crates/gunnar-ui-server/src/api/twin.rs
//! **The twin**: its heartbeat as the primary sees it, and the one-step restore
//! from it.
//!
//! # The heartbeat panel (`GET /api/twin`, operator)
//!
//! gunnar's `Admin.Health` carries one `TwinHeartbeat` row per twin (every
//! member of group `twin`, and every caller that reported as one) and, on a
//! twin, its primary's last answer. [`render`] is the one place that reads
//! that message. The state word is gunnar's (`ok` / `behind` / `silent`),
//! never re-derived here: the thresholds live in gunnar, and a console that
//! decided "silent" by its own clock would be a second opinion.
//!
//! # Restore from the twin (`GET` + `POST /api/restore`, admin) — the Twin panel
//!
//! When the primary's data disk died and the box was re-imaged, it comes up
//! FRESH: the bootstrap window open, no identity, no key. The twin still holds
//! every transaction. **[Review twin]** (`GET`) is gunnar-repair's
//! `review_twin` — health, head vs primary, last heartbeat, what a recharge
//! would lose; **[Recharge from twin]** (`POST`) is gunnar-repair's
//! `recharge_from_twin`, the ONE pump behind this button, gunnar's boot check
//! and a pulled monetize order (5.4 resilience, 2026-10-10). This module holds
//! only the adapters over this console's channels ([`UiTwin`], [`UiMain`],
//! [`UiFence`]) and the console's own gates:
//!
//! * **admin** (`ADMIN_SURFACES`) and **stepped up** ([`step_up`]): a
//! user-verified login from the last ten minutes, the break-glass token, or
//! — the main being unable to log anyone in — an admin key the TWIN vouches
//! for ([`STEP_UP_HEADER`], [`twin_admin`]; the public `/twin-door/restore`);
//! * **the confirmation**: `GET` answers the phrase to type
//! (`restore <head_seq> from twin`); the POST carries it back, so a twin
//! whose head moved in between asks again;
//! * **a live main is refused** unless `override_live_main` (its own phrase),
//! which FENCES it first (`fence_primary`: standby, roster cleared) and
//! still streams nothing into it — it is re-laid, then recharged.
//!
//! **How the console reaches the twin**: `GUNNAR_UI_TWIN` names its control
//! plane, and the channel is signed with this console's own key
//! ([`Gunnar::sibling`]) — the twin's identity table is the primary's,
//! replayed, so it knows this key. `Bootstrap.Restore` on the fresh box is
//! signed with this console's key, bound to `Bootstrap.Restore` (h-v1-6a).
//!
//! **A fetch that fails FAILS the restore stream** (gunnar-repair's poison
//! codec) instead of cancelling it: the fresh box sees a broken stream and
//! stays restoring (resumable by running the restore again).
use std::time::Duration;
use axum::extract::State;
use axum::Json;
use serde::{Deserialize, Serialize};
use gunnar_repair::pump::Fault;
use crate::api::{ApiError, ApiResult};
use crate::app::AppState;
use crate::auth::middleware::Caller;
use crate::grpc::{Fail, Gunnar};
use crate::pb;
/// The environment variable naming the twin's control plane (`host:port`).
pub const TWIN_ENV: &str = "GUNNAR_UI_TWIN";
/// How long the twin gets to answer `Mirror.LogHead` before it counts as
/// unreachable.
const HEAD_WAIT: Duration = Duration::from_secs(10);
/// The configured twin: its address as written, and a channel to it signed
/// with the console's key.
#[derive(Clone)]
pub struct Twin {
pub address: String,
pub gunnar: Gunnar,
}
impl Twin {
/// A lazy channel to `address`, signed like `console`.
pub fn connect(console: &Gunnar, address: &str) -> Result<Twin, String> {
let address = address.trim().to_owned();
let gunnar = console.sibling(&address)?;
Ok(Twin { address, gunnar })
}
}
/// One twin, as the panel draws it.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TwinRow {
pub principal: String,
pub address: String,
/// `None` = never heard from since gunnar started.
pub last_seen_unix_ms: Option<i64>,
pub seen_ago_ms: u64,
pub applied_seq: u64,
pub head_seq: u64,
pub lag_entries: u64,
pub disk_total_bytes: u64,
pub disk_used_bytes: u64,
/// `None` when the twin did not measure its volume (total 0).
pub fill_permille: Option<u32>,
pub state: String,
pub why: String,
pub silent_after_ms: u64,
}
/// A twin's view of its primary.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct PrimaryRow {
pub address: String,
pub last_answer_unix_ms: Option<i64>,
pub answer_ago_ms: u64,
pub primary_head_seq: u64,
pub applied_seq: u64,
pub state: String,
pub why: String,
}
/// `GET /api/twin`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TwinPanel {
pub twins: Vec<TwinRow>,
pub primary: Option<PrimaryRow>,
/// Non-empty when gunnar could not answer (an older gunnar, or one that
/// does not serve `Admin.Health`), in words. Empty lists then mean
/// "not measured", never "no twin".
pub unanswered: String,
/// The twin this console is configured to reach (`GUNNAR_UI_TWIN`), empty
/// when none is. CONFIGURATION, not a measurement, and the only thing on
/// this panel that does not come from gunnar: the restore's eligibility is
/// an admin's read, so without this an OPERATOR looking at a box whose twin
/// is dead cannot tell "this box has no twin" from "the twin has never
/// been heard from". It says where, never that anything answered there.
pub configured_twin: String,
}
/// The one reader of gunnar's message.
pub fn render(h: pb::HealthResponse) -> TwinPanel {
TwinPanel {
twins: h
.twins
.into_iter()
.map(|t| TwinRow {
principal: t.principal,
address: t.address,
last_seen_unix_ms: (t.last_seen_unix_ms > 0).then_some(t.last_seen_unix_ms),
seen_ago_ms: t.seen_ago_ms,
applied_seq: t.applied_seq,
head_seq: t.head_seq,
lag_entries: t.lag_entries,
disk_total_bytes: t.disk_total_bytes,
disk_used_bytes: t.disk_used_bytes,
fill_permille: (t.disk_total_bytes > 0).then_some(t.fill_permille),
state: t.state,
why: t.why,
silent_after_ms: t.silent_after_ms,
})
.collect(),
primary: h.primary.map(|p| PrimaryRow {
address: p.address,
last_answer_unix_ms: (p.last_answer_unix_ms > 0).then_some(p.last_answer_unix_ms),
answer_ago_ms: p.answer_ago_ms,
primary_head_seq: p.primary_head_seq,
applied_seq: p.applied_seq,
state: p.state,
why: p.why,
}),
unanswered: String::new(),
configured_twin: String::new(),
}
}
/// `GET /api/twin` — the heartbeat rows. Not cached: a panel that shows a
/// twin as `ok` for a cache's lifetime after it went silent is the failure
/// this panel exists to show.
pub async fn panel(
reader: crate::api::Reader,
State(state): State<AppState>,
) -> ApiResult<TwinPanel> {
// Where the restore would read from, carried on the operator's read so a
// dead twin does not look like no twin. Never merged into the rows: a
// configured address is not a heartbeat.
let configured_twin = state.twin.as_ref().map(|t| t.address.clone()).unwrap_or_default();
match reader.gunnar.admin().health(pb::HealthRequest {}).await {
Ok(r) => Ok(Json(TwinPanel { configured_twin, ..render(r.into_inner()) })),
Err(s) if s.code() == tonic::Code::Unimplemented => Ok(Json(TwinPanel {
unanswered: format!("this Gunnar does not answer Admin.Health: {}", s.message()),
configured_twin,
..TwinPanel::default()
})),
Err(s) => Err(ApiError::from(s)),
}
}
gunnar_repair::wire!(crate::pb);
/// `GET /api/restore` — the panel's [Review twin].
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Eligibility {
/// Draw the action.
pub eligible: bool,
/// The configured twin, empty when `GUNNAR_UI_TWIN` is unset.
pub twin_address: String,
pub twin_reachable: bool,
pub twin_first_seq: u64,
pub twin_head_seq: u64,
pub window_open: bool,
pub principals: u64,
pub keys: u64,
/// The phrase the POST must carry back; empty unless eligible, or refused
/// only for a reason the POST may waive (`stale_ok`, `accept_lost`, `primary_fenced`).
pub confirm: String,
/// Why not, in words; empty when eligible.
pub why: String,
/// gunnar-repair's refusal code (`main_live`, `not_fresh`, `stale_source`,
/// `would_lose`, `unreachable`, `no_twin`, …); empty when eligible.
pub code: String,
/// The main answers and is NOT fresh: a recharge needs `override_live_main`,
/// which FENCES it first (and still streams nothing into a live box).
pub main_live: bool,
/// The phrase an override must carry back; empty unless `main_live`.
pub override_confirm: String,
/// **gunnar-repair's `review_twin`**: health, head vs primary, last
/// heartbeat, what a recharge would lose. `None` with no twin configured.
pub review: Option<gunnar_repair::TwinReview>,
}
/// The phrase that confirms a restore of a twin holding `head_seq`.
pub fn confirm_phrase(head_seq: u64) -> String {
format!("restore {head_seq} from twin")
}
/// The phrase that confirms fencing a LIVE main before a restore from a twin
/// holding `head_seq`.
pub fn override_phrase(head_seq: u64) -> String {
format!("fence main and restore {head_seq} from twin")
}
/// Refusals a POST may waive by saying so (`stale_ok`, `accept_lost`,
/// `primary_fenced`): the confirmation is drawn for them, the person reads the
/// review first. `not_fenced` (R-13): the twin still hears a primary on ANOTHER
/// machine — the person fences or stops it, then says so.
const WAIVABLE: &[&str] = &["stale_source", "would_lose", "not_fenced"];
/// **The panel's read of gunnar-repair's review** — `status` the main's
/// `Bootstrap.Status` (`Err` = it did not answer), `twin` the twin's
/// `Mirror.LogHead`. The decision is `gunnar_repair::review_twin`'s; this
/// only lays it out for the page.
pub fn eligible(
twin_address: &str,
status: Result<&gunnar_repair::MainStatus, &str>,
twin: Result<&gunnar_repair::TwinHead, &str>,
now_unix_ms: i64,
) -> Eligibility {
let mut e = Eligibility { twin_address: twin_address.to_owned(), ..Eligibility::default() };
if let Ok(t) = twin {
e.twin_reachable = true;
e.twin_first_seq = t.first_seq;
e.twin_head_seq = t.head_seq;
}
if let Ok(s) = status {
e.window_open = s.window_open;
e.principals = s.principals;
e.keys = s.keys;
e.main_live = !s.window_open && !s.restoring;
}
if twin_address.is_empty() {
e.twin_reachable = false;
e.code = "no_twin".to_owned();
e.why = format!("no twin is configured ({TWIN_ENV} is unset)");
return e;
}
let review = gunnar_repair::review_twin(twin_address, twin, status.ok(), None, now_unix_ms);
(e.code, e.why) = match status {
Err(why) => ("unreachable".to_owned(), format!("this Gunnar did not answer Bootstrap.Status: {why}")),
Ok(_) if review.recharge_possible => (String::new(), String::new()),
Ok(_) => (review.refusal.clone(), review.verdict.clone()),
};
e.eligible = e.why.is_empty();
if e.eligible || WAIVABLE.contains(&e.code.as_str()) {
e.confirm = confirm_phrase(e.twin_head_seq);
}
if e.main_live && e.twin_reachable && e.twin_head_seq > 0 {
e.override_confirm = override_phrase(e.twin_head_seq);
}
e.review = Some(review);
e
}
fn now_unix_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
.unwrap_or(0)
}
// ── gunnar-repair's host traits over this console's channels ──────────────
/// **The twin**, over `Mirror` signed with this console's key.
struct UiTwin<'a>(&'a Twin);
impl gunnar_repair::pump::Twin for UiTwin<'_> {
type Chunk = pb::LogChunk;
type Stream = gunnar_repair::grpc::TonicStream<pb::LogChunk>;
fn addr(&self) -> String {
self.0.address.clone()
}
async fn head(&self) -> Result<gunnar_repair::TwinHead, Fault> {
use gunnar_repair::AsTwinHead;
match tokio::time::timeout(HEAD_WAIT, self.0.gunnar.mirror().log_head(pb::LogHeadRequest::default())).await {
Err(_) => Err(Fault::new("deadlineexceeded", format!("no answer within {} s", HEAD_WAIT.as_secs()))),
Ok(Err(s)) => Err(gunnar_repair::grpc::fault(&s)),
Ok(Ok(r)) => Ok(r.into_inner().twin_head()),
}
}
async fn snapshot(&self, from_item: u64, from_offset: u64, expect_id: &str) -> Result<Self::Stream, Fault> {
let req = pb::SnapshotFetchRequest { from_item, from_offset, expect_id: expect_id.to_owned() };
let s = self.0.gunnar.mirror().snapshot_fetch(req).await.map_err(|s| gunnar_repair::grpc::fault(&s))?;
Ok(gunnar_repair::grpc::TonicStream(s.into_inner()))
}
async fn log(&self, from: u64, to: u64, from_offset: u64) -> Result<Self::Stream, Fault> {
let req = pb::LogFetchRequest { from, to, max_bytes: FETCH_MAX_BYTES, from_offset };
let s = self.0.gunnar.mirror().log_fetch(req).await.map_err(|s| gunnar_repair::grpc::fault(&s))?;
Ok(gunnar_repair::grpc::TonicStream(s.into_inner()))
}
}
/// The byte ceiling of one `LogFetch` (gunnar's CLI's, 32 MiB).
const FETCH_MAX_BYTES: u64 = 32 << 20;
/// **The main** this console fronts: `Bootstrap.Status`, and `Bootstrap.Restore`
/// signed with the console's key bound to exactly that method (h-v1-6a: the
/// fresh box takes a restore only from a key it was FED, and the console's is
/// one — `offer.<front>.console_pubkey`).
struct UiMain<'a>(&'a Gunnar);
impl gunnar_repair::pump::Main<pb::LogChunk> for UiMain<'_> {
type Sink = gunnar_repair::grpc::TonicRestore<pb::LogChunk, pb::RestoreResponse>;
async fn status(&self) -> Result<gunnar_repair::MainStatus, Fault> {
use gunnar_repair::AsMainStatus;
let r = self.0.bootstrap().status(pb::BootstrapStatusRequest {}).await.map_err(|s| gunnar_repair::grpc::fault(&s))?;
Ok(r.into_inner().main_status())
}
async fn open_restore(&self) -> Result<Self::Sink, Fault> {
let header = self.0.credential_for("Bootstrap.Restore").map_err(|e| Fault::new("internal", format!("cannot sign the restore: {e}")))?;
let value: tonic::metadata::MetadataValue<tonic::metadata::Ascii> =
header.parse().map_err(|_| Fault::new("internal", "the signed credential is not a valid header value"))?;
let mut grpc = self.0.bootstrap_grpc();
Ok(gunnar_repair::grpc::restore_stream(move |body, codec| async move {
grpc.ready().await.map_err(|e| tonic::Status::unavailable(format!("the fresh box's channel was not ready: {e}")))?;
let mut request = tonic::Request::new(body);
request.metadata_mut().insert(crate::signer::CREDENTIAL_METADATA_KEY, value);
grpc.client_streaming(request, http::uri::PathAndQuery::from_static(gunnar_repair::grpc::RESTORE_PATH), codec).await
}))
}
}
/// **The main, to fence** (`Mirror.SetRole(STANDBY)` then
/// `Mirror.DemoteReplication`), signed with the console's key.
struct UiFence<'a>(&'a Gunnar);
impl gunnar_repair::fence::Fenceable for UiFence<'_> {
fn addr(&self) -> String {
"this console's Gunnar".to_owned()
}
async fn set_standby(&self) -> Result<String, Fault> {
let r = self
.0
.mirror()
.set_role(pb::SetRoleRequest { role: pb::ReplicationRole::Standby as i32 })
.await
.map_err(|s| gunnar_repair::grpc::fault(&s))?;
Ok(match pb::ReplicationRole::try_from(r.into_inner().role) {
Ok(pb::ReplicationRole::Standby) => gunnar_repair::fence::STANDBY.to_owned(),
Ok(other) => other.as_str_name().to_ascii_lowercase(),
Err(_) => "unknown".to_owned(),
})
}
async fn demote(&self) -> Result<usize, Fault> {
let r = self.0.mirror().demote_replication(pb::DemoteReplicationRequest {}).await.map_err(|s| gunnar_repair::grpc::fault(&s))?;
Ok(r.into_inner().targets.len())
}
}
async fn measure(main: &Gunnar, twin: Option<&Twin>) -> Eligibility {
use gunnar_repair::pump::{Main as _, Twin as _};
let status = UiMain(main).status().await.map_err(|f| f.message);
let (address, head) = match twin {
None => (String::new(), Err("not configured".to_owned())),
Some(t) => (t.address.clone(), UiTwin(t).head().await.map_err(|f| f.message)),
};
eligible(&address, status.as_ref().map_err(String::as_str), head.as_ref().map_err(String::as_str), now_unix_ms())
}
/// `GET /api/restore` — [Review twin]: may the action be drawn, with which
/// phrase, and gunnar-repair's review of the twin.
pub async fn eligibility(State(state): State<AppState>) -> ApiResult<Eligibility> {
Ok(Json(measure(&state.gunnar, state.twin.as_ref()).await))
}
/// The POST body.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct RestoreBody {
/// Must equal the `confirm` phrase `GET /api/restore` answered (or its
/// `override_confirm` with `override_live_main`).
#[serde(default)]
pub confirm: String,
/// Take a twin whose log starts late with no snapshot to bridge the gap.
#[serde(default)]
pub accept_lost: bool,
/// Restore a twin that cannot say it is current.
#[serde(default)]
pub stale_ok: bool,
/// The main answers as LIVE: fence it (standby, roster cleared) first.
#[serde(default)]
pub override_live_main: bool,
/// The person says the twin's OLD primary is fenced or stopped (waives
/// `not_fenced`; a fenced primary still answers its twin, so it cannot tell).
#[serde(default)]
pub primary_fenced: bool,
}
/// What the restore did — `Bootstrap.Restore`'s answer, and the fence when one ran.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Restored {
pub head_seq: u64,
pub applied: u64,
pub lost: u64,
pub restart_required: bool,
pub detail: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fenced: Option<gunnar_repair::fence::Fenced>,
}
fn refuse(code: &str, message: impl Into<String>) -> ApiError {
ApiError(Fail { code: code.to_owned(), message: message.into() })
}
/// The header the twin door's step-up rides in: a `gunnar-sshsig-v2`
/// credential bound to `Access.WhoAmI`, signed by the person's own key.
pub const STEP_UP_HEADER: &str = "x-gunnar-step-up";
/// **Step-up for a recharge** (5.4): a session whose user-verified login is
/// older than `STEP_UP_WINDOW_SECS` logs in again — or, when the main cannot
/// log anyone in (its data set is gone), proves an admin key to the TWIN
/// ([`STEP_UP_HEADER`]). The break-glass token and the loopback-only open
/// console (no credential configured at all) pass, as on every admin verb.
pub(crate) async fn step_up(state: &AppState, caller: Option<&Caller>, headers: &http::HeaderMap) -> Result<String, ApiError> {
match caller {
None => Ok("the open console".to_owned()),
Some(Caller::AdminToken) => Ok("the admin token".to_owned()),
Some(Caller::Operator(s)) if s.stepped_up(crate::auth::now_secs()) => Ok(s.user.clone()),
Some(Caller::Operator(_)) if headers.contains_key(STEP_UP_HEADER) => twin_admin(state, headers).await,
// The machine-readable "a fresh touch is needed" (`auth::STEP_UP_CODE`):
// the web console asks the key in place and sends the recharge again.
Some(Caller::Operator(_)) => Err(refuse(
crate::auth::STEP_UP_CODE,
format!(
"{} With the main down, prove an admin key to the twin in {STEP_UP_HEADER} instead.",
crate::auth::step_up_words()
),
)),
}
}
/// **The twin vouches for an admin key** — the twin door's login: the
/// credential in [`STEP_UP_HEADER`] is relayed to the twin's `Access.WhoAmI`,
/// and only a principal the twin names as a SERVER admin passes. Works with
/// the main dead: the twin holds the main's identity table, replicated.
pub(crate) async fn twin_admin(state: &AppState, headers: &http::HeaderMap) -> Result<String, ApiError> {
let Some(twin) = state.twin.as_ref() else {
return Err(refuse("failedprecondition", format!("no twin is configured ({TWIN_ENV} is unset), so no twin can vouch for a key")));
};
let Some(credential) = headers.get(STEP_UP_HEADER).and_then(|v| v.to_str().ok()) else {
return Err(refuse("permissiondenied", format!("the twin door needs {STEP_UP_HEADER}: a gunnar-sshsig-v2 credential for Access.WhoAmI")));
};
let who = twin
.gunnar
.who_am_i_as(credential)
.await
.map_err(|s| refuse("permissiondenied", format!("the twin at {} did not vouch for this key: {}", twin.address, s.message())))?;
let principal = who.principal.map(|p| p.id).unwrap_or_default();
if who.server_level != pb::GrantLevel::Admin as i32 || principal.is_empty() {
return Err(refuse(
"permissiondenied",
format!("the twin at {} says this key ({}) is not a server admin's", twin.address, if principal.is_empty() { "nobody" } else { &principal }),
));
}
Ok(principal)
}
/// `POST /api/restore` — [Recharge from twin]: admin (`ADMIN_SURFACES`),
/// stepped up, the confirmation phrase, and the review measured again at the
/// moment of the act. ONE path: `gunnar_repair::recharge_from_twin`.
pub async fn restore(
State(state): State<AppState>,
caller: Option<axum::Extension<Caller>>,
headers: http::HeaderMap,
Json(body): Json<RestoreBody>,
) -> ApiResult<Restored> {
let who = step_up(&state, caller.as_deref(), &headers).await?;
recharge(&state, &who, body).await.map(Json)
}
/// `GET /twin-door/restore` — the review, for a person whose main cannot log
/// them in: the twin vouches for their key ([`twin_admin`]).
pub async fn door_review(State(state): State<AppState>, headers: http::HeaderMap) -> ApiResult<Eligibility> {
twin_admin(&state, &headers).await?;
Ok(Json(measure(&state.gunnar, state.twin.as_ref()).await))
}
/// `POST /twin-door/restore` — [`restore`], stepped up by the twin.
pub async fn door_restore(State(state): State<AppState>, headers: http::HeaderMap, Json(body): Json<RestoreBody>) -> ApiResult<Restored> {
let who = twin_admin(&state, &headers).await?;
recharge(&state, &who, body).await.map(Json)
}
async fn recharge(state: &AppState, who: &str, body: RestoreBody) -> Result<Restored, ApiError> {
let typed = body.confirm.trim();
if typed.is_empty() {
return Err(refuse(
"invalidargument",
"a restore needs its confirmation: read GET /api/restore and send back the phrase it names",
));
}
let e = measure(&state.gunnar, state.twin.as_ref()).await;
let Some(twin) = state.twin.as_ref() else {
return Err(refuse("failedprecondition", e.why));
};
if e.main_live && body.override_live_main {
if typed != e.override_confirm || e.override_confirm.is_empty() {
return Err(refuse(
"invalidargument",
format!("fencing a live main needs \"{}\": the twin's head moved, or it was mistyped", e.override_confirm),
));
}
// FENCE FIRST: never two writable boxes. A fenced main is still not
// fresh, so nothing streams into it — it is re-laid, then recharged.
let fenced = gunnar_repair::fence::fence_primary(&UiFence(&state.gunnar), 0)
.await
.map_err(|r| refuse(if r.code == gunnar_repair::RefusalCode::Unreachable { "unavailable" } else { "failedprecondition" }, r.to_string()))?;
tracing::info!(target: crate::auth::AUDIT_TARGET, event = "twin.fence", by = %who, twin = %twin.address, "the live main was fenced from the Twin panel");
return Err(refuse(
"failedprecondition",
format!(
"{}. The main is fenced and still holds its data, so nothing was streamed into it: re-lay it (a fresh box), then recharge that from the twin",
fenced.detail
),
));
}
let waived = (e.code == "stale_source" && body.stale_ok)
|| (e.code == "would_lose" && body.accept_lost)
|| (e.code == "not_fenced" && body.primary_fenced);
if !e.eligible && !waived {
return Err(refuse("failedprecondition", e.why));
}
if typed != e.confirm {
return Err(refuse(
"invalidargument",
format!("the confirmation does not match \"{}\": the twin's head moved, or it was mistyped", e.confirm),
));
}
let opts = gunnar_repair::RechargeOpts {
accept_lost: body.accept_lost,
stale_ok: body.stale_ok,
primary_fenced: body.primary_fenced,
..Default::default()
};
tracing::info!(target: crate::auth::AUDIT_TARGET, event = "twin.recharge", by = %who, twin = %twin.address, "recharge from the twin started from the Twin panel");
let receipt = gunnar_repair::recharge_from_twin(&UiTwin(twin), &UiMain(&state.gunnar), &opts, &|| false)
.await
.map_err(|r| refuse(if r.code == gunnar_repair::RefusalCode::Unreachable { "unavailable" } else { "failedprecondition" }, r.to_string()))?;
match receipt.phase {
gunnar_repair::Phase::Done => Ok(Restored {
head_seq: receipt.head_seq,
applied: receipt.applied,
lost: receipt.lost,
restart_required: receipt.needs_restart,
detail: receipt.detail,
fenced: None,
}),
_ => Err(refuse(
"aborted",
format!(
"{}; the fresh box was left restoring — run the restore again",
if receipt.why.is_empty() { receipt.detail } else { receipt.why }
),
)),
}
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use tonic::{Request, Response, Status, Streaming};
use tower::ServiceExt;
use super::*;
use gunnar_repair::{AsMainStatus, AsTwinHead};
/// A re-laid box: window open, nobody enrolled, and new enough to take a
/// snapshot resumably.
fn fresh() -> pb::BootstrapStatusResponse {
pb::BootstrapStatusResponse { window_open: true, restore_snapshot_resumable: true, ..Default::default() }
}
/// The fake twin's log identity: 128 bits in hex, the shape
/// `gunnar_server::twin::LOG_ID` mints.
const LOG_ID: &str = "5f1e0c9a7d3b42e8a61f0b9c2d4e7a13";
/// A twin whose primary went silent while the two were LEVEL (the shape
/// of a dead main), holding `first..=n` — and, when its log starts late,
/// the snapshot of everything below (named on its head).
fn head(first: u64, n: u64) -> pb::LogHeadResponse {
pb::LogHeadResponse {
head_seq: n,
first_seq: first,
log_id: LOG_ID.to_owned(),
twin: true,
applied_seq: n,
primary_head_seq: n,
state: "unreachable".to_owned(),
last_answer_unix_ms: 1_700_000_000_000,
last_level_unix_ms: 1_700_000_000_000,
answer_ago_ms: 30_000,
snapshot_at: first.saturating_sub(1),
snapshot_bytes: if first > 1 { 3 } else { 0 },
..Default::default()
}
}
/// A twin that still HEARS its primary, level with it two seconds ago: the
/// old primary is up (a re-laid main elsewhere, the old box still standing).
fn live_head(first: u64, n: u64) -> pb::LogHeadResponse {
pb::LogHeadResponse {
state: "following".to_owned(),
caught_up: true,
last_answer_unix_ms: 1_700_000_028_000,
last_level_unix_ms: 1_700_000_028_000,
answer_ago_ms: 2_000,
..head(first, n)
}
}
fn judge(addr: &str, status: Result<&pb::BootstrapStatusResponse, &str>, twin: Result<&pb::LogHeadResponse, &str>) -> Eligibility {
let status = status.map(AsMainStatus::main_status);
let twin = twin.map(AsTwinHead::twin_head);
eligible(addr, status.as_ref().map_err(|e| *e), twin.as_ref().map_err(|e| *e), 1_700_000_030_000)
}
#[test]
fn a_fresh_box_with_a_reachable_twin_is_eligible_and_nothing_else_is() {
// A primary gone SILENT is no proof it is dead (fence before promote): refused
// `not_fenced`, and waivable — the confirmation is drawn for the person who fenced it.
let e = judge("10.13.8.122:50051", Ok(&fresh()), Ok(&head(1, 9)));
assert!(!e.eligible && e.code == "not_fenced", "{}: {}", e.code, e.why);
assert_eq!((e.twin_head_seq, e.confirm.as_str()), (9, "restore 9 from twin"));
let review = e.review.expect("the review rides along");
assert_eq!((review.health, review.would_lose.entries), (Some(gunnar_repair::Health::PrimarySilent), 0));
// A primary that came back EMPTY (re-laid at its address) holds none of the old
// writes: no fence is needed, and the fresh box is eligible outright.
let relaid = pb::LogHeadResponse { primary_head_seq: 0, state: "diverged".into(), last_answer_unix_ms: 1_700_000_028_000, answer_ago_ms: 2_000, ..head(1, 9) };
let e = judge("10.13.8.122:50051", Ok(&fresh()), Ok(&relaid));
assert!(e.eligible, "{}: {}", e.code, e.why);
assert_eq!((e.twin_head_seq, e.confirm.as_str()), (9, "restore 9 from twin"));
let enrolled = pb::BootstrapStatusResponse { window_open: false, bootstrapped: true, principals: 1, keys: 1, ..Default::default() };
let e = judge("t:1", Ok(&enrolled), Ok(&head(1, 9)));
assert!(!e.eligible && e.code == "main_live" && e.main_live && e.confirm.is_empty(), "{}: {}", e.code, e.why);
assert_eq!(e.override_confirm, "fence main and restore 9 from twin");
// Keys present with the window still open (identities predating the record).
let keyed = pb::BootstrapStatusResponse { window_open: true, keys: 1, ..Default::default() };
let e = judge("t:1", Ok(&keyed), Ok(&head(1, 9)));
assert!(!e.eligible && e.code == "not_fresh", "{}: {}", e.code, e.why);
let e = judge("t:1", Ok(&fresh()), Err("connection refused"));
assert!(!e.eligible && !e.twin_reachable && e.why.contains("did not answer Mirror.LogHead"), "{}", e.why);
let e = judge("", Ok(&fresh()), Ok(&head(1, 9)));
assert!(!e.eligible && e.why.contains(TWIN_ENV), "{}", e.why);
assert!(!judge("t:1", Err("unavailable"), Ok(&head(1, 9))).eligible);
let e = judge("t:1", Ok(&fresh()), Ok(&head(0, 0)));
assert!(!e.eligible && e.code == "empty_twin", "an empty twin restores nothing: {}", e.code);
// A box answering as a primary is no copy: the review and the POST say the same.
let primary = pb::LogHeadResponse { twin: false, ..head(1, 9) };
assert_eq!(judge("t:1", Ok(&fresh()), Ok(&primary)).code, "not_a_twin");
// A twin that cannot say how current it is: refused, but waivable by the POST. It
// never heard its primary, so the fence is asked first (`stale_ok` is the next gate).
let idle = pb::LogHeadResponse { state: "idle".into(), last_answer_unix_ms: 0, ..head(1, 9) };
let e = judge("t:1", Ok(&fresh()), Ok(&idle));
assert!(!e.eligible && e.code == "not_fenced" && e.confirm == "restore 9 from twin", "{}: {}", e.code, e.why);
}
/// **R-13 on the panel**: a twin that still hears its primary is refused
/// `not_fenced` — and the person may waive it, by saying the old primary is
/// fenced or stopped, after reading why.
#[test]
fn a_twin_that_still_hears_its_primary_is_refused_not_fenced_and_waivable() {
let e = judge("t:1", Ok(&fresh()), Ok(&live_head(1, 9)));
assert!(!e.eligible && e.code == "not_fenced", "{}: {}", e.code, e.why);
assert!(e.why.contains("fence"), "the refusal says what to do: {}", e.why);
assert_eq!(e.confirm, "restore 9 from twin", "waivable: the phrase is drawn");
assert!(!e.main_live && e.override_confirm.is_empty(), "THIS box is fresh: nothing here to override");
}
fn beat(principal: &str, state: &str, lag: u64, seen: i64) -> pb::TwinHeartbeat {
pb::TwinHeartbeat {
principal: principal.into(),
address: format!("10.13.8.122:{}", 40000 + lag),
last_seen_unix_ms: seen,
seen_ago_ms: 2_000,
applied_seq: 100 - lag,
head_seq: 100,
lag_entries: lag,
disk_total_bytes: if seen > 0 { 2 << 30 } else { 0 },
disk_used_bytes: if seen > 0 { 1 << 30 } else { 0 },
fill_permille: if seen > 0 { 500 } else { 0 },
state: state.into(),
why: if state == "ok" { String::new() } else { format!("{state} by gunnar's measure") },
silent_after_ms: 30_000,
..Default::default()
}
}
#[test]
fn render_carries_gunnars_words_and_absent_is_not_zero() {
let h = pb::HealthResponse {
twins: vec![beat("twin", "ok", 2, 1_700_000_000_000), beat("old-twin", "silent", 0, 0)],
primary: None,
..Default::default()
};
let p = render(h);
assert_eq!(p.twins.len(), 2);
assert_eq!((p.twins[0].fill_permille, p.twins[0].lag_entries, p.twins[0].state.as_str()), (Some(500), 2, "ok"));
assert_eq!((p.twins[1].last_seen_unix_ms, p.twins[1].fill_permille), (None, None));
assert!(p.primary.is_none() && p.unanswered.is_empty());
}
// ── fakes: a primary (Admin + Bootstrap) and a twin (Mirror) ───────────
/// A twin whose log is `first..=n`, each entry two chunks; `fail_after`
/// makes a later fetch stream break after that many chunks.
#[derive(Clone)]
struct FakeTwin {
first: u64,
n: u64,
fail_after: Option<usize>,
/// Its primary still answers (R-13): [`live_head`] instead of [`head`].
live: bool,
}
type Chunks = std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<pb::LogChunk, Status>> + Send + 'static>>;
#[tonic::async_trait]
impl pb::mirror_server::Mirror for FakeTwin {
async fn log_head(&self, _: Request<pb::LogHeadRequest>) -> Result<Response<pb::LogHeadResponse>, Status> {
Ok(Response::new(if self.live { live_head(self.first, self.n) } else { head(self.first, self.n) }))
}
async fn log_fetch(&self, r: Request<pb::LogFetchRequest>) -> Result<Response<Chunks>, Status> {
let r = r.into_inner();
assert!(r.from >= self.first, "asked for {} below the twin's first {}", r.from, self.first);
// Two entries per fetch, so the pump has to ask more than once.
let to = r.to.min(r.from + 1);
let mut out: Vec<Result<pb::LogChunk, Status>> = Vec::new();
for seq in r.from..=to {
for part in 0..2u8 {
out.push(Ok(pb::LogChunk {
seq,
kind: 1,
meta: if part == 0 { format!("m{seq}").into_bytes() } else { Vec::new() },
blob_len: 2,
// Not announced: this fixture streams a LOG, not a
// restore, so there is no size to check a set against.
restore_bytes: 0,
blob: vec![part],
last: part == 1,
restore_head: 0,
// What gunnar's `Mirror.LogFetch` sends: EMPTY. The id
// is `LogHead.log_id`, and `Bootstrap.Restore` is the
// only reader of this field (`gunnar-grpc`
// `service/mirror.rs` builds every chunk with
// `log_id: String::new()`). The pump stamps it.
log_id: String::new(),
// An older sender's chunk (gunnar's restore-large-entry fields, 2026-10-07,
// absent): each entry from its first byte, the length check alone.
offset: None,
entry_crc: None,
// restore-after-snapshot's fields (2026-10-07), absent from a log chunk.
snapshot_item: None,
snapshot_id: String::new(),
}));
}
}
if let Some(k) = self.fail_after {
if r.from > self.first {
out.truncate(k);
out.push(Err(Status::unavailable("the twin's disk went away")));
}
}
Ok(Response::new(Box::pin(tokio_stream::iter(out))))
}
/// Its snapshot, positioned (a twin as new as the pump): three items,
/// then the `SnapshotAt` terminator.
async fn snapshot_fetch(&self, _: Request<pb::SnapshotFetchRequest>) -> Result<Response<Chunks>, Status> {
let piece = |i: u64, kind: u32| pb::LogChunk {
seq: 0,
kind,
last: true,
blob: vec![1],
offset: Some(0),
snapshot_item: Some(i),
snapshot_id: "snap-1".to_owned(),
..Default::default()
};
let mut items: Vec<Result<pb::LogChunk, Status>> = (0..3u8).map(|i| Ok(piece(u64::from(i), 90 + u32::from(i)))).collect();
items.push(Ok(piece(3, gunnar_repair::KIND_SNAPSHOT_AT)));
Ok(Response::new(Box::pin(tokio_stream::iter(items))))
}
}
/// The primary: `Admin.Health` rows, `Bootstrap.Status`, and a
/// `Bootstrap.Restore` that records every chunk.
#[derive(Clone, Default)]
struct FakeMain {
twins: Vec<pb::TwinHeartbeat>,
status: pb::BootstrapStatusResponse,
got: Arc<Mutex<Vec<(u64, u32, bool)>>>,
/// The `log_id` of every chunk the box was sent, in order.
log_ids: Arc<Mutex<Vec<String>>>,
finished: Arc<Mutex<Option<Result<(), String>>>>,
/// The credential the restore stream carried (h-v1-6a).
credential: Arc<Mutex<Option<String>>>,
/// The fence's calls, in order.
fence: Arc<Mutex<Vec<&'static str>>>,
}
#[tonic::async_trait]
impl pb::mirror_server::Mirror for FakeMain {
async fn set_role(&self, r: Request<pb::SetRoleRequest>) -> Result<Response<pb::ReplicationStatusResponse>, Status> {
assert_eq!(r.into_inner().role, pb::ReplicationRole::Standby as i32);
self.fence.lock().unwrap().push("set_role");
Ok(Response::new(pb::ReplicationStatusResponse { role: pb::ReplicationRole::Standby as i32, ..Default::default() }))
}
async fn demote_replication(&self, _: Request<pb::DemoteReplicationRequest>) -> Result<Response<pb::ReplicationStatusResponse>, Status> {
self.fence.lock().unwrap().push("demote");
Ok(Response::new(pb::ReplicationStatusResponse { role: pb::ReplicationRole::Standby as i32, ..Default::default() }))
}
}
/// The admin's own key, and a reader's: what the fake twin's `Access.WhoAmI` knows.
fn person_key(seed: u8) -> ssh_key::PrivateKey {
ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&[seed; 32]))
}
const ADMIN_SEED: u8 = 5;
const READER_SEED: u8 = 6;
/// The twin answers `WhoAmI` for the credential it is handed — gunnar's
/// check, ported (`signer::verify_header`), over the two keys it holds.
#[tonic::async_trait]
impl pb::access_server::Access for FakeTwin {
async fn who_am_i(&self, r: Request<pb::WhoAmIRequest>) -> Result<Response<pb::WhoAmIResponse>, Status> {
let header = r.metadata().get(crate::signer::CREDENTIAL_METADATA_KEY).and_then(|v| v.to_str().ok()).unwrap_or_default().to_owned();
if !header.contains(" Access.WhoAmI ") {
return Err(Status::unauthenticated("a credential for another method"));
}
for (seed, id, level) in [(ADMIN_SEED, "root", pb::GrantLevel::Admin), (READER_SEED, "reader", pb::GrantLevel::Read)] {
if crate::signer::verify_header(&person_key(seed).public_key().to_openssh().unwrap(), &header) {
return Ok(Response::new(pb::WhoAmIResponse {
principal: Some(pb::Principal { id: id.to_owned(), ..Default::default() }),
server_level: level as i32,
..Default::default()
}));
}
}
Err(Status::unauthenticated("no key of this twin signed that"))
}
}
fn step_up_for(seed: u8) -> String {
crate::signer::Signer::new(person_key(seed)).unwrap().credential_for_now("Access.WhoAmI").unwrap()
}
#[tonic::async_trait]
impl pb::admin_server::Admin for FakeMain {
async fn health(&self, _: Request<pb::HealthRequest>) -> Result<Response<pb::HealthResponse>, Status> {
Ok(Response::new(pb::HealthResponse { serving: true, twins: self.twins.clone(), ..Default::default() }))
}
}
#[tonic::async_trait]
impl pb::bootstrap_server::Bootstrap for FakeMain {
async fn status(&self, _: Request<pb::BootstrapStatusRequest>) -> Result<Response<pb::BootstrapStatusResponse>, Status> {
Ok(Response::new(self.status.clone()))
}
async fn restore(&self, r: Request<Streaming<pb::LogChunk>>) -> Result<Response<pb::RestoreResponse>, Status> {
*self.credential.lock().unwrap() = r.metadata().get(crate::signer::CREDENTIAL_METADATA_KEY).and_then(|v| v.to_str().ok()).map(str::to_owned);
let mut s = r.into_inner();
let mut applied = 0;
loop {
match s.message().await {
Ok(Some(c)) => {
if c.last && c.seq > 0 {
applied += 1;
}
self.got.lock().unwrap().push((c.seq, c.kind, c.last));
self.log_ids.lock().unwrap().push(c.log_id);
}
Ok(None) => break,
Err(e) => {
*self.finished.lock().unwrap() = Some(Err(e.message().to_owned()));
return Err(Status::aborted("broken stream; still restoring"));
}
}
}
*self.finished.lock().unwrap() = Some(Ok(()));
let head = self.got.lock().unwrap().last().map(|x| x.0).unwrap_or(0);
Ok(Response::new(pb::RestoreResponse { head_seq: head, applied, lost: 0, restart_required: true, detail: "restart gunnar".into() }))
}
}
async fn listen() -> (tokio::net::TcpListener, String) {
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let a = l.local_addr().unwrap();
(l, format!("http://{a}"))
}
async fn serve_twin(t: FakeTwin) -> String {
let (l, at) = listen().await;
tokio::spawn(
tonic::transport::Server::builder()
.add_service(pb::mirror_server::MirrorServer::new(t.clone()))
.add_service(pb::access_server::AccessServer::new(t))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(l)),
);
at
}
async fn serve_main(m: FakeMain) -> String {
let (l, at) = listen().await;
tokio::spawn(
tonic::transport::Server::builder()
.add_service(pb::bootstrap_server::BootstrapServer::new(m.clone()))
.add_service(pb::mirror_server::MirrorServer::new(m.clone()))
.add_service(pb::admin_server::AdminServer::new(m))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(l)),
);
at
}
/// An address nothing listens on.
async fn nowhere() -> String {
let (l, at) = listen().await;
drop(l);
at
}
fn console_key(endpoint: &str) -> Gunnar {
let signer = crate::signer::Signer::new(ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&[9u8; 32]))).unwrap();
Gunnar::connect_lazy(endpoint, Arc::new(signer)).unwrap()
}
/// The console's router in front of `main_at`, with `twin_at` configured
/// (or no twin). Open console: no passkeys, so the ladder is the anonymous
/// suite's business (`anonymous::the_twin_panel_is_an_operators_and_the_restore_an_admins`).
fn router(main_at: &str, twin_at: Option<&str>) -> axum::Router {
crate::serve::router(app_state(main_at, twin_at))
}
fn app_state(main_at: &str, twin_at: Option<&str>) -> AppState {
let gunnar = console_key(main_at);
let twin = twin_at.map(|a| Twin::connect(&gunnar, a).unwrap());
let store = crate::state::StateStore::open(tempfile::Builder::new().suffix(".redb").tempfile().unwrap().into_temp_path().keep().unwrap()).unwrap();
Arc::new(crate::app::App {
webauthn: None,
#[cfg(feature = "metered")]
monetize: None,
admin_token: None,
events: crate::sse::EventHub::detached(),
gunnar,
twin,
guest: None,
reviewers: Arc::new(crate::reviewer::ReviewerKeys::new(main_at.to_owned(), store.clone())),
tenants: Arc::new(crate::tenant::TenantKeys::new(main_at.to_owned(), store)),
forges: Default::default(),
imports: Default::default(),
forge_apps: Default::default(),
ui_dir: None,
caches: crate::api::Caches::default(),
objects: Default::default(),
readmes: Default::default(),
})
}
async fn call(app: &axum::Router, method: http::Method, uri: &str, body: Option<&str>) -> (http::StatusCode, serde_json::Value) {
call_with(app, method, uri, body, None).await
}
async fn call_with(app: &axum::Router, method: http::Method, uri: &str, body: Option<&str>, step_up: Option<&str>) -> (http::StatusCode, serde_json::Value) {
let mut b = http::Request::builder().method(method).uri(uri);
if let Some(credential) = step_up {
b = b.header(STEP_UP_HEADER, credential);
}
if body.is_some() {
b = b.header("content-type", "application/json");
}
let request = b.body(axum::body::Body::from(body.unwrap_or("").to_owned())).unwrap();
let response = app.clone().oneshot(request).await.unwrap();
let status = response.status();
let bytes = axum::body::to_bytes(response.into_body(), 1 << 20).await.unwrap();
(status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null))
}
fn post_restore(confirm: &str) -> String {
serde_json::json!({ "confirm": confirm }).to_string()
}
/// The POST of a person who says the twin's old primary is fenced or stopped.
/// **Fence before promote** (gunnar-repair, owner 2026-10-10): a twin that merely
/// cannot hear its primary is no proof that primary is dead, so every recharge from
/// a silent-primary twin ([`head`]) is `not_fenced` until somebody says so.
fn post_fenced_restore(confirm: &str) -> String {
serde_json::json!({ "confirm": confirm, "primary_fenced": true }).to_string()
}
// ── GET /api/twin ──────────────────────────────────────────────────────
#[tokio::test]
async fn the_panel_carries_each_twins_state_ok_behind_and_silent() {
let main = FakeMain {
twins: vec![beat("twin-a", "ok", 0, 1_700_000_000_000), beat("twin-b", "behind", 40, 1_700_000_000_000), beat("twin-c", "silent", 0, 0)],
..FakeMain::default()
};
let app = router(&serve_main(main).await, None);
let (status, json) = call(&app, http::Method::GET, "/api/twin", None).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
let rows = json["twins"].as_array().expect("rows");
let states: Vec<&str> = rows.iter().map(|r| r["state"].as_str().unwrap()).collect();
assert_eq!(states, ["ok", "behind", "silent"]);
assert_eq!(rows[1]["lag_entries"], 40);
assert_eq!(rows[0]["fill_permille"], 500);
assert_eq!(rows[0]["address"], "10.13.8.122:40000");
assert!(rows[2]["last_seen_unix_ms"].is_null() && rows[2]["fill_permille"].is_null(), "{json}");
assert_eq!(json["unanswered"], "");
assert_eq!(json["configured_twin"], "", "no twin was configured, and the panel said one was");
}
/// **A twin that is down is not the absence of a twin.** gunnar reports no
/// row for a twin that never polled and is in no group; the operator's
/// panel still has to say where the second copy is meant to be, so the
/// configured address rides along — beside the rows, never inside them.
#[tokio::test]
async fn the_panel_names_the_configured_twin_even_when_gunnar_reports_no_row() {
let main_at = serve_main(FakeMain::default()).await;
let app = router(&main_at, Some("10.13.8.122:50051"));
let (status, json) = call(&app, http::Method::GET, "/api/twin", None).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
assert_eq!(json["twins"].as_array().map(Vec::len), Some(0), "a row was invented for a twin nobody heard from");
assert_eq!(json["configured_twin"], "10.13.8.122:50051");
}
// ── /api/restore ───────────────────────────────────────────────────────
#[tokio::test]
async fn a_restore_without_its_confirmation_is_refused_before_anything_is_dialled() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let got = main.got.clone();
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await));
for body in ["{}", &post_restore(""), &post_restore(" ")] {
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(body)).await;
assert_eq!(status, http::StatusCode::BAD_REQUEST, "{json}");
assert_eq!(json["code"], "invalidargument", "{json}");
}
// A phrase that is not the one GET answered.
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_fenced_restore("restore 2 from twin"))).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::BAD_REQUEST, Some("invalidargument")), "{json}");
assert!(got.lock().unwrap().is_empty(), "a chunk reached the box without the confirmation");
}
#[tokio::test]
async fn a_box_that_is_not_fresh_draws_no_action_and_refuses_the_restore() {
let enrolled = pb::BootstrapStatusResponse { bootstrapped: true, principals: 2, keys: 3, ..Default::default() };
let main = FakeMain { status: enrolled, ..FakeMain::default() };
let got = main.got.clone();
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await));
let (status, json) = call(&app, http::Method::GET, "/api/restore", None).await;
assert_eq!(status, http::StatusCode::OK);
assert_eq!((json["eligible"].as_bool(), json["confirm"].as_str()), (Some(false), Some("")), "{json}");
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_restore("restore 3 from twin"))).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::BAD_REQUEST, Some("failedprecondition")), "{json}");
assert!(json["message"].as_str().unwrap().contains("main_live"), "{json}");
assert!(got.lock().unwrap().is_empty());
}
#[tokio::test]
async fn an_unreachable_or_unconfigured_twin_refuses_the_restore() {
let main_at = serve_main(FakeMain { status: fresh(), ..FakeMain::default() }).await;
let away = nowhere().await;
for (twin, words) in [(Some(away.as_str()), "did not answer Mirror.LogHead"), (None, TWIN_ENV)] {
let app = router(&main_at, twin);
let (_, json) = call(&app, http::Method::GET, "/api/restore", None).await;
assert_eq!((json["eligible"].as_bool(), json["twin_reachable"].as_bool()), (Some(false), Some(false)), "{json}");
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_restore("restore 3 from twin"))).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::BAD_REQUEST, Some("failedprecondition")), "{json}");
assert!(json["message"].as_str().unwrap().contains(words), "{json}");
}
}
#[tokio::test]
async fn the_confirmed_restore_streams_the_twin_into_the_fresh_box_and_answers_gunnars_numbers() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let (got, finished) = (main.got.clone(), main.finished.clone());
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 5, fail_after: None, live: false }).await));
let (status, json) = call(&app, http::Method::GET, "/api/restore", None).await;
assert_eq!(status, http::StatusCode::OK);
assert_eq!((json["eligible"].as_bool(), json["code"].as_str()), (Some(false), Some("not_fenced")), "{json}");
let confirm = json["confirm"].as_str().unwrap().to_owned();
assert_eq!(confirm, "restore 5 from twin");
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_fenced_restore(&confirm))).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
assert_eq!((json["head_seq"].as_u64(), json["applied"].as_u64(), json["lost"].as_u64()), (Some(5), Some(5), Some(0)));
assert_eq!((json["restart_required"].as_bool(), json["detail"].as_str()), (Some(true), Some("restart gunnar")));
let want: Vec<(u64, u32, bool)> = (1..=5u64).flat_map(|s| [(s, 1, false), (s, 1, true)]).collect();
assert_eq!(*got.lock().unwrap(), want, "every chunk, in order, exactly once");
assert_eq!(*finished.lock().unwrap(), Some(Ok(())));
}
/// **The restore stream carries the console's credential, bound to
/// `Bootstrap.Restore`** (h-v1-6a): a fresh box takes a restore only from a
/// key it was fed, and the console's is one (`offer.<front>.console_pubkey`).
#[tokio::test]
async fn the_restore_is_signed_with_the_consoles_key_for_bootstrap_restore() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let credential = main.credential.clone();
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await));
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_fenced_restore("restore 3 from twin"))).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
let header = credential.lock().unwrap().clone().expect("the restore stream carries a credential");
assert!(header.contains(" Bootstrap.Restore "), "bound to Bootstrap.Restore: {header}");
let key = ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&[9u8; 32]));
let public = key.public_key().to_openssh().unwrap();
assert!(crate::signer::verify_header(&public, &header), "signed by the console's key: {header}");
}
// ── the pump ───────────────────────────────────────────────────────────
#[tokio::test]
async fn a_twin_whose_log_starts_late_sends_its_snapshot_first() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let got = main.got.clone();
let app = router(&serve_main(main.clone()).await, Some(&serve_twin(FakeTwin { first: 7, n: 9, fail_after: None, live: false }).await));
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_fenced_restore("restore 9 from twin"))).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
assert_eq!((json["head_seq"].as_u64(), json["applied"].as_u64()), (Some(9), Some(3)), "{json}");
let got = got.lock().unwrap().clone();
let mut want = vec![(0, 90, true), (0, 91, true), (0, 92, true), (0, gunnar_repair::KIND_SNAPSHOT_AT, true)];
want.extend((7..=9u64).flat_map(|s| [(s, 1, false), (s, 1, true)]));
assert_eq!(got, want, "the snapshot, then the log from its first entry");
let ids = main.log_ids.lock().unwrap().clone();
assert_eq!(ids.len(), got.len());
assert!(
ids.iter().all(|id| id == LOG_ID),
"every chunk — snapshot and log — must carry the twin's LogHead.log_id, or the \
restored box mints a new identity and its twin stops following it: {ids:?}"
);
}
#[tokio::test]
async fn a_fetch_that_breaks_aborts_the_restore_instead_of_ending_the_stream_cleanly() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let app = router(&serve_main(main.clone()).await, Some(&serve_twin(FakeTwin { first: 1, n: 6, fail_after: Some(1), live: false }).await));
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_fenced_restore("restore 6 from twin"))).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::BAD_GATEWAY, Some("aborted")), "{json}");
let err = json["message"].as_str().unwrap_or_default();
assert!(err.contains("left restoring"), "{err}");
for _ in 0..50 {
if main.finished.lock().unwrap().is_some() {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let finished = main.finished.lock().unwrap().clone();
assert!(!matches!(finished, Some(Ok(()))), "the fresh box saw a CLEAN end of a short stream: {finished:?}");
}
// ── 5.4: step-up, the twin door, a live main ──────────────────────────
/// **A recharge needs a user-verified login from the last ten minutes.**
/// An admin whose login is older is sent to log in again; the token and
/// the open console pass, as on every admin verb.
#[tokio::test]
async fn a_recharge_needs_a_stepped_up_login() {
let main_at = serve_main(FakeMain { status: fresh(), ..FakeMain::default() }).await;
let twin_at = serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await;
let state = app_state(&main_at, Some(&twin_at));
let now = crate::auth::now_secs();
let none = http::HeaderMap::new();
let stale = Caller::Operator(crate::auth::Session::for_test("h", crate::role::Role::Admin, now - 3_600, true));
let err = step_up(&state, Some(&stale), &none).await.expect_err("an hour-old login is not a step-up");
assert_eq!(err.0.code, crate::auth::STEP_UP_CODE, "freshness, not a role refusal");
assert_eq!(err.0.http_status(), http::StatusCode::FORBIDDEN);
assert!(err.0.message.contains("fresh touch"), "{}", err.0.message);
let no_uv = Caller::Operator(crate::auth::Session::for_test("h", crate::role::Role::Admin, now, false));
assert!(step_up(&state, Some(&no_uv), &none).await.is_err(), "presence alone is not a step-up");
let fresh_uv = Caller::Operator(crate::auth::Session::for_test("h", crate::role::Role::Admin, now, true));
assert_eq!(step_up(&state, Some(&fresh_uv), &none).await.unwrap(), "h");
assert!(step_up(&state, Some(&Caller::AdminToken), &none).await.is_ok());
assert!(step_up(&state, None, &none).await.is_ok(), "the open console is loopback-only and has no ladder");
// With main down, an old session steps up through the twin.
let mut headers = http::HeaderMap::new();
headers.insert(STEP_UP_HEADER, step_up_for(ADMIN_SEED).parse().unwrap());
assert_eq!(step_up(&state, Some(&stale), &headers).await.unwrap(), "root");
}
/// **The twin door**: no session at all (the main cannot log anyone in);
/// only an admin key the TWIN names passes, and then the recharge is the
/// same as the panel's.
#[tokio::test]
async fn the_twin_door_recharges_only_for_a_key_the_twin_names_a_server_admin() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let got = main.got.clone();
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await));
let body = post_fenced_restore("restore 3 from twin");
let (status, json) = call_with(&app, http::Method::POST, "/twin-door/restore", Some(&body), None).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::FORBIDDEN, Some("permissiondenied")), "{json}");
let (status, json) = call_with(&app, http::Method::POST, "/twin-door/restore", Some(&body), Some(&step_up_for(READER_SEED))).await;
assert_eq!(status, http::StatusCode::FORBIDDEN, "{json}");
assert!(json["message"].as_str().unwrap().contains("not a server admin"), "{json}");
let (status, json) = call_with(&app, http::Method::POST, "/twin-door/restore", Some(&body), Some(&step_up_for(9))).await;
assert_eq!(status, http::StatusCode::FORBIDDEN, "a key the twin does not hold: {json}");
assert!(got.lock().unwrap().is_empty(), "a chunk moved before the twin vouched for an admin");
let (status, json) = call_with(&app, http::Method::GET, "/twin-door/restore", None, Some(&step_up_for(ADMIN_SEED))).await;
assert_eq!((status, json["confirm"].as_str()), (http::StatusCode::OK, Some("restore 3 from twin")), "{json}");
let (status, json) = call_with(&app, http::Method::POST, "/twin-door/restore", Some(&body), Some(&step_up_for(ADMIN_SEED))).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
assert_eq!(json["head_seq"].as_u64(), Some(3), "{json}");
assert_eq!(got.lock().unwrap().len(), 6, "three entries, two chunks each");
}
/// **The POST waives `not_fenced` only when it says `primary_fenced`**, and
/// then recharges with that said: without it nothing is streamed.
#[tokio::test]
async fn a_recharge_beside_a_live_primary_streams_only_once_the_person_says_it_is_fenced() {
let main = FakeMain { status: fresh(), ..FakeMain::default() };
let (got, fence) = (main.got.clone(), main.fence.clone());
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: true }).await));
let (_, json) = call(&app, http::Method::GET, "/api/restore", None).await;
assert_eq!((json["code"].as_str(), json["confirm"].as_str()), (Some("not_fenced"), Some("restore 3 from twin")), "{json}");
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_restore("restore 3 from twin"))).await;
assert_eq!((status, json["code"].as_str()), (http::StatusCode::BAD_REQUEST, Some("failedprecondition")), "{json}");
assert!(json["message"].as_str().unwrap().contains("fence"), "{json}");
assert!(got.lock().unwrap().is_empty(), "streamed beside a live primary");
let said = serde_json::json!({ "confirm": "restore 3 from twin", "primary_fenced": true }).to_string();
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&said)).await;
assert_eq!(status, http::StatusCode::OK, "{json}");
assert_eq!((json["head_seq"].as_u64(), json["applied"].as_u64()), (Some(3), Some(3)), "{json}");
assert!(fence.lock().unwrap().is_empty(), "the panel fenced THIS box: the old primary is another machine");
}
/// **A live main is refused; the override FENCES it first** — standby,
/// then the roster cleared — and still streams nothing into it.
#[tokio::test]
async fn a_live_main_is_refused_and_the_override_fences_it_before_anything_else() {
let live = pb::BootstrapStatusResponse { window_open: false, bootstrapped: true, principals: 2, keys: 3, ..Default::default() };
let main = FakeMain { status: live, ..FakeMain::default() };
let (got, fence) = (main.got.clone(), main.fence.clone());
let app = router(&serve_main(main).await, Some(&serve_twin(FakeTwin { first: 1, n: 3, fail_after: None, live: false }).await));
let (_, json) = call(&app, http::Method::GET, "/api/restore", None).await;
assert_eq!((json["main_live"].as_bool(), json["code"].as_str()), (Some(true), Some("main_live")), "{json}");
assert_eq!(json["override_confirm"], "fence main and restore 3 from twin");
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&post_restore("restore 3 from twin"))).await;
assert_eq!(status, http::StatusCode::BAD_REQUEST, "{json}");
assert!(fence.lock().unwrap().is_empty(), "fenced without the override");
let wrong = serde_json::json!({ "confirm": "restore 3 from twin", "override_live_main": true }).to_string();
let (status, _) = call(&app, http::Method::POST, "/api/restore", Some(&wrong)).await;
assert_eq!(status, http::StatusCode::BAD_REQUEST);
assert!(fence.lock().unwrap().is_empty(), "fenced on the plain phrase");
let over = serde_json::json!({ "confirm": "fence main and restore 3 from twin", "override_live_main": true }).to_string();
let (status, json) = call(&app, http::Method::POST, "/api/restore", Some(&over)).await;
assert_eq!(status, http::StatusCode::BAD_REQUEST, "{json}");
let message = json["message"].as_str().unwrap();
assert!(message.contains("fenced") && message.contains("re-lay"), "{message}");
assert_eq!(*fence.lock().unwrap(), vec!["set_role", "demote"], "standby first, then the roster");
assert!(got.lock().unwrap().is_empty(), "a chunk was streamed into a live main");
}
}