crates/gunnar-front/src/operator_door.rs
//! **The operator's verbs on vm1's door** (the plugin door: mTLS, the operator's
//! certificate from the estate's installer CA, `CN=installer`, only):
//!
//! * **`GET /backup/stream` — the forge, for an off-site copy**
//! (h-forge-offsite, 2026-09-28). The TWIN's history as a fresh box's
//! `Bootstrap.Restore` takes it: its `Mirror.SnapshotFetch` when its log
//! starts after 1, then `Mirror.LogFetch` from its first entry to its head,
//! every chunk stamped with the head (`restore_head`), the log's size
//! (`restore_bytes`) and its id — the ONE pump
//! (`gunnar_repair::recharge_from_twin`, as `gunnar control twin replay`,
//! the console's button and a monetize order), written to frames. Read with the console's key (an admin on both boxes). No twin in
//! the feed: the primary's own log. The body is frames, `[u32 BE len][LogChunk
//! protobuf]`, and one zero-length frame at the END: a stream that breaks
//! never carries the end marker, so a copy of it is never kept. Headers:
//! `x-gunnar-source`, `-log-id`, `-first-seq`, `-head-seq`, `-bytes`.
//! * **`GET /twin/health` — the twin's lag and load, for its autoscale**
//! (2026-10-05). The PRIMARY's `Admin.Health` heartbeat rows, read with the
//! console's key over the overlay (WireGuard), as JSON: every twin row and
//! the one `reading` a private estate's monetize sizes the twin by — the
//! heard twin furthest behind (its lag, its own cores busy/total), or the
//! word `silent` when none is heard. Read only; nothing changes.
//! * **`PUT /slot/<name>` — the operator's bytes into a SEALED SLOT on the
//! primary** (F4, the owner 2026-10-09). The body is stored with
//! `Sealed.Put` under this front's certify key (the machine principal
//! `front-certify`, which then OWNS the slot: nobody else reads or writes
//! it); sealed under the primary's KEK, carried by its twin, snapshots and a
//! recharge. Answers `STORED` only once the primary said so — never the
//! value. How a sold estate keeps its installer material after the hand-off
//! ([`INSTALLER_MATERIAL_SLOT`]), off this front's disk.
//! * **`POST /feed/<key>` — rotate one fed credential in place**
//! (h-admin-token / h-rotation-owed, 2026-09-28). `{"value": "…"}` replaces
//! ONE of [`ROTATABLE`] in the front's kept feed (sealed on disk,
//! h-feed-at-rest); the console swaps its bearer on its next look (2 s). No
//! restart. Answers `ROTATED` (or `UNCHANGED`), never the value.
use std::time::Duration;
use axum::body::Body;
use axum::extract::{Path as UrlPath, State};
use axum::http::{HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use crate::plugin_door::{Conn, NOT_OPERATOR};
/// The feed keys an operator may rotate through the door, and nothing else: the
/// console's bearer. The front holds no Stripe or UpCloud secret (ONE monetize,
/// owner 13:35Z 2026-10-10).
pub const ROTATABLE: &[&str] = &[crate::console::FEED_ADMIN_TOKEN];
/// The twin's lag and load, for its autoscale.
pub const TWIN_HEALTH_PATH: &str = "/twin/health";
/// **The slot a sold estate's installer material is kept in** (F4): the CA, the
/// installer's certificate and link keys, the door ledger, the plan and the
/// seeds — what re-admits an upgraded or recharged box and renews certificates.
pub const INSTALLER_MATERIAL_SLOT: &str = "installer-material";
/// How long one fetch may take to START answering.
const FETCH_WAIT: Duration = Duration::from_secs(30);
/// The routes, merged into the plugin door's router.
pub(crate) fn routes() -> Router<Conn> {
Router::new()
.route("/backup/stream", get(get_backup))
.route(TWIN_HEALTH_PATH, get(get_twin_health))
.route("/feed/{key}", post(post_feed))
.route("/slot/{name}", axum::routing::put(put_slot))
.route(REBUILD_LOOK_PATH, get(get_rebuild_look))
}
/// **`GET /rebuild/look` — what the main-rebuild guard reads off the estate** (owner
/// 2026-10-12: a dead main is rebuilt from its twin, never the twin promoted). The main's
/// `Bootstrap.Status` (or that it did not answer) and the twin's own review, both over the
/// overlay with the console's key: the operator's kit judges with them
/// (`gunnar_repair::rebuild::judge`) before anything is destroyed. A read: nothing changes.
pub const REBUILD_LOOK_PATH: &str = "/rebuild/look";
/// How long the look waits for each box.
const LOOK_WAIT: Duration = Duration::from_secs(10);
/// Pure: **the look's answer** — the main as the guard reads it ([`gunnar_repair::rebuild::main_of`])
/// and the twin's review (`null` when the front knows no twin).
pub fn rebuild_look_json(
main_at: &str,
main: Result<&gunnar_repair::MainStatus, &str>,
twin: Option<&gunnar_repair::TwinReview>,
) -> serde_json::Value {
let (seen, head) = gunnar_repair::rebuild::main_of(main);
serde_json::json!({
"main": { "at": main_at, "seen": seen, "head": head, "why": main.err().unwrap_or("") },
"twin": twin,
})
}
async fn get_rebuild_look(State(c): State<Conn>) -> Response {
use gunnar_repair::pump::{Main as _, Twin as _};
if let Some(r) = not_operator(&c, "the rebuild's look") {
return r;
}
let (_, target) = (c.ctx.now)();
let Some(target) = target else {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"rebuild-no-primary",
"no primary is known yet (the installer has not fed this front)",
);
};
let key = match tokio::fs::read_to_string(c.ctx.root.join(crate::console::IDENTITY_REL))
.await
.map_err(|e| e.to_string())
.and_then(|pem| {
gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).map_err(|e| e.to_string())
}) {
Ok(k) => k,
Err(e) => {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"rebuild-no-console-key",
format!("the console's key cannot be read: {e}"),
)
}
};
let main = match mirror_channel(&target.control) {
Ok(channel) => {
let m = gunnar_repair::proto::ProtoMain {
addr: target.control.clone(),
channel,
key: key.clone(),
};
match tokio::time::timeout(LOOK_WAIT, m.status()).await {
Ok(Ok(s)) => Ok(s),
Ok(Err(f)) => Err(format!("Bootstrap.Status: {f}")),
Err(_) => Err(format!("no answer within {} s", LOOK_WAIT.as_secs())),
}
}
Err(e) => Err(e),
};
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let twin = match &target.twin {
None => None,
Some(t) => {
let head = match mirror_channel(t) {
Ok(channel) => {
let tw = gunnar_repair::proto::ProtoTwin {
addr: t.clone(),
channel,
key: key.clone(),
};
match tokio::time::timeout(LOOK_WAIT, tw.head()).await {
Ok(Ok(h)) => Ok(h),
Ok(Err(f)) => Err(format!("Mirror.LogHead: {f}")),
Err(_) => Err(format!("no answer within {} s", LOOK_WAIT.as_secs())),
}
}
Err(e) => Err(e),
};
let base = head
.as_ref()
.ok()
.and_then(gunnar_repair::TwinBase::from_head);
Some(gunnar_repair::review_twin(
t,
head.as_ref().map_err(String::as_str),
main.as_ref().ok(),
base,
now_ms,
))
}
};
tracing::info!(main = %target.control, twin = ?target.twin, "plugin door: the operator looks at the estate before a main rebuild");
(
StatusCode::OK,
Json(rebuild_look_json(
&target.control,
main.as_ref().map_err(String::as_str),
twin.as_ref(),
)),
)
.into_response()
}
fn refused(status: StatusCode, clause: &str, why: impl Into<String>) -> Response {
let why = why.into();
(status, Json(serde_json::json!({"outcome": format!("REFUSED [{clause}]"), "line": format!("REFUSED [{clause}] {why}")}))).into_response()
}
fn not_operator(c: &Conn, what: &str) -> Option<Response> {
c.operator.as_ref().err().map(|cn| refused(StatusCode::FORBIDDEN, NOT_OPERATOR, format!("the client certificate is `{cn}`, not the operator's (`{}`): {what} is the operator's only", nordisk_pairing::mtls::INSTALLER_CN)))
}
/// **The twin's lag and load as the door answers them** — pure: every row
/// of the primary's heartbeat, and the `reading` — the HEARD twin (`ok` or
/// `behind`) furthest behind, with its own cores; `silent` when no twin is
/// heard (nothing was read off one); `none` when the primary names no twin.
pub fn twin_health_json(h: &gunnar_proto::HealthResponse) -> serde_json::Value {
let rows: Vec<serde_json::Value> = h
.twins
.iter()
.map(|t| {
serde_json::json!({
"principal": t.principal,
"state": t.state,
"twin_state": t.twin_state,
"lag_entries": t.lag_entries,
"seen_ago_ms": t.seen_ago_ms,
"cpu_millicores_busy": t.cpu_millicores_busy,
"cpu_millicores_total": t.cpu_millicores_total,
})
})
.collect();
let heard = h
.twins
.iter()
.filter(|t| t.state == "ok" || t.state == "behind")
.max_by_key(|t| t.lag_entries);
let reading = match heard {
Some(t) => serde_json::json!({
"principal": t.principal,
"state": t.state,
"lag_entries": t.lag_entries,
"cpu_millicores_busy": t.cpu_millicores_busy,
"cpu_millicores_total": t.cpu_millicores_total,
}),
None => {
serde_json::json!({ "state": if h.twins.is_empty() { "none" } else { "silent" }, "lag_entries": 0 })
}
};
serde_json::json!({ "twins": rows, "reading": reading })
}
/// `GET /twin/health` — the operator's, read only. (The shop never reads it: ONE monetize
/// reads a sold estate's twin off the boxes' own Fleet reports, G12; F4 removed the shop's
/// health-reader certificate.)
async fn get_twin_health(State(c): State<Conn>) -> Response {
if let Some(r) = not_operator(&c, "the twin's health") {
return r;
}
let (_, target) = (c.ctx.now)();
let Some(target) = target else {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"twin-health-no-primary",
"no primary is known yet (the installer has not fed this front)",
);
};
let key = match tokio::fs::read_to_string(c.ctx.root.join(crate::console::IDENTITY_REL))
.await
.map_err(|e| e.to_string())
.and_then(|pem| {
gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).map_err(|e| e.to_string())
}) {
Ok(k) => k,
Err(e) => {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"twin-health-no-console-key",
format!("the console's key cannot be read: {e}"),
)
}
};
let channel = match tonic::transport::Channel::from_shared(format!("http://{}", target.control))
{
Ok(ch) => ch.connect_timeout(FETCH_WAIT).connect_lazy(),
Err(e) => {
return refused(
StatusCode::INTERNAL_SERVER_ERROR,
"twin-health-address",
format!("{}: {e}", target.control),
)
}
};
let mut admin = gunnar_proto::admin_client::AdminClient::new(
tonic::service::interceptor::InterceptedService::new(
channel,
gunnar_grpc::client_signer::CredentialSigner,
),
);
match tokio::time::timeout(
FETCH_WAIT,
admin.health(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::HealthRequest::default(),
)),
)
.await
{
Ok(Ok(h)) => (StatusCode::OK, Json(twin_health_json(&h.into_inner()))).into_response(),
Ok(Err(s)) => refused(
StatusCode::BAD_GATEWAY,
"twin-health-primary",
format!(
"the primary at {} answered Admin.Health: {}",
target.control,
s.message()
),
),
Err(_) => refused(
StatusCode::GATEWAY_TIMEOUT,
"twin-health-primary",
format!(
"the primary at {} did not answer Admin.Health within {} s",
target.control,
FETCH_WAIT.as_secs()
),
),
}
}
/// **May the off-site copy be read from the twin?** — pure. `Ok(())`: yes;
/// `Err(why)`: read it from the PRIMARY instead. MEASURED on the release-3
/// mock 2026-09-29: a twin whose data set was mid-growth (state
/// `maintenance`: drained, its set unmounted) answered LogHead `first=1` and
/// then `Mirror.LogFetch from 1: txlog: No such file or directory` — the copy
/// broke and none was kept. So the twin is used only when it answers LogHead,
/// is `following` (or idle: a twin not polled since start, whose log is on
/// disk), and ONE entry at its first sequence reads back (`probe`).
pub fn twin_verdict(
head: Result<&gunnar_proto::LogHeadResponse, &str>,
probe: Result<(), &str>,
) -> Result<(), String> {
let head = head.map_err(|e| format!("the twin did not answer Mirror.LogHead ({e})"))?;
if !matches!(head.state.as_str(), "following" | "idle" | "") {
return Err(format!(
"the twin is `{}` ({}), not following",
head.state,
if head.why.is_empty() {
"no reason given"
} else {
head.why.as_str()
}
));
}
probe.map_err(|e| {
format!(
"the twin's log does not read back at seq {} ({e})",
head.first_seq.max(1)
)
})
}
/// **May the copy be streamed from the PRIMARY?** — pure. It is the source when
/// the twin was passed over ([`twin_verdict`]'s `why`, empty when no twin is
/// fed); `probe` is ONE entry at its first sequence read back. `Err((clause,
/// line))`: a set held for its own growth (unmounted) or otherwise unreadable —
/// refused BY NAME before a byte is streamed, both reasons named (defect B,
/// release-3 mock 2026-09-29: a copy that broke mid-way on `txlog: No such file`).
pub fn primary_verdict(
twin_why: &str,
at: &str,
probe: Result<(), &str>,
) -> Result<(), (&'static str, String)> {
probe.map_err(|e| {
let twin = if twin_why.is_empty() { "not fed" } else { twin_why };
("backup-source-unreadable", format!("no source can be read now — the twin: {twin}; the primary at {at}: {e}. Nothing was streamed; try again once the growth has mounted the set back"))
})
}
/// One frame: `[u32 BE len][bytes]`.
pub fn frame(bytes: &[u8]) -> Vec<u8> {
let mut v = Vec::with_capacity(bytes.len() + 4);
v.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
v.extend_from_slice(bytes);
v
}
/// The end marker: a zero-length frame.
pub const END: [u8; 4] = [0, 0, 0, 0];
/// A signed control channel to the source box at `at`.
/// A signed channel to a box's control plane over the tunnel (the console's lanes).
pub(crate) fn control_channel(
at: &str,
) -> Result<gunnar_grpc::client_signer::ControlChannel, String> {
mirror_channel(at)
}
fn mirror_channel(at: &str) -> Result<gunnar_grpc::client_signer::ControlChannel, String> {
let channel = tonic::transport::Channel::from_shared(format!("http://{at}"))
.map_err(|e| format!("{at}: {e}"))?
.connect_timeout(FETCH_WAIT)
.connect_lazy();
Ok(tonic::service::interceptor::InterceptedService::new(
channel,
gunnar_grpc::client_signer::CredentialSigner,
))
}
/// Pure: how the copy asks the pump. The door already chose its source ([`twin_verdict`],
/// [`primary_verdict`]): a primary is allowed, and a twin's age is the copy's to keep, not
/// to refuse (the copy says its source in `x-gunnar-source`).
/// The frames take no client write, so a copy beside a live primary opens no second
/// writer: `primary_fenced` (the fence rule is for a recharge into a box that serves).
pub fn copy_opts(source: &str) -> gunnar_repair::RechargeOpts {
gunnar_repair::RechargeOpts {
stale_ok: true,
allow_primary_source: source == "primary",
primary_fenced: true,
..gunnar_repair::RechargeOpts::default()
}
}
/// Pure: the box the copy is written for — a fresh one that takes the source's snapshot
/// from its start (so a log that begins after 1 streams its base first).
pub fn fresh_box() -> gunnar_repair::MainStatus {
gunnar_repair::MainStatus {
window_open: true,
restore_snapshot_resumable: true,
..gunnar_repair::MainStatus::default()
}
}
/// **The copy as the pump's main**: every chunk the pump stamps (head, size, log id)
/// becomes one frame, a whole stream ends with [`END`], a broken one with an error.
struct FrameCopy(tokio::sync::mpsc::Sender<Result<bytes::Bytes, std::io::Error>>);
struct FrameSink {
tx: tokio::sync::mpsc::Sender<Result<bytes::Bytes, std::io::Error>>,
through: u64,
applied: u64,
/// The log entries written so far (stinky row 17): a hole ends the copy, never an END.
run: Option<LogRun>,
}
impl FrameSink {
async fn broke(&self, why: String) -> gunnar_repair::Fault {
let _ = self.tx.send(Err(std::io::Error::other(why.clone()))).await;
gunnar_repair::Fault::new("data_loss", why)
}
}
/// **The copy's source, with every message awaited under [`MESSAGE_WAIT`]** (stinky row 17):
/// a source that goes silent mid-stream ends the copy instead of holding it open for ever.
struct Watched(gunnar_repair::proto::ProtoTwin);
struct WatchedStream(<gunnar_repair::proto::ProtoTwin as gunnar_repair::pump::Twin>::Stream);
impl gunnar_repair::pump::ChunkStream for WatchedStream {
type Chunk = gunnar_proto::LogChunk;
async fn next(&mut self) -> Option<Result<gunnar_proto::LogChunk, gunnar_repair::Fault>> {
match within(MESSAGE_WAIT, "the source's stream", async {
Ok(self.0.next().await)
})
.await
{
Ok(next) => next,
Err(why) => Some(Err(gunnar_repair::Fault::new("deadline_exceeded", why))),
}
}
}
impl gunnar_repair::pump::Twin for Watched {
type Chunk = gunnar_proto::LogChunk;
type Stream = WatchedStream;
fn addr(&self) -> String {
self.0.addr()
}
async fn head(&self) -> Result<gunnar_repair::TwinHead, gunnar_repair::Fault> {
self.0.head().await
}
async fn snapshot(
&self,
from_item: u64,
from_offset: u64,
expect_id: &str,
) -> Result<WatchedStream, gunnar_repair::Fault> {
Ok(WatchedStream(
self.0.snapshot(from_item, from_offset, expect_id).await?,
))
}
async fn log(
&self,
from: u64,
to: u64,
from_offset: u64,
) -> Result<WatchedStream, gunnar_repair::Fault> {
Ok(WatchedStream(self.0.log(from, to, from_offset).await?))
}
}
impl gunnar_repair::pump::Main<gunnar_proto::LogChunk> for FrameCopy {
type Sink = FrameSink;
async fn status(&self) -> Result<gunnar_repair::MainStatus, gunnar_repair::Fault> {
Ok(fresh_box())
}
async fn open_restore(&self) -> Result<FrameSink, gunnar_repair::Fault> {
Ok(FrameSink {
tx: self.0.clone(),
through: 0,
applied: 0,
run: None,
})
}
}
impl gunnar_repair::pump::RestoreSink<gunnar_proto::LogChunk> for FrameSink {
async fn send(&mut self, chunk: gunnar_proto::LogChunk) -> Result<(), gunnar_repair::Fault> {
use prost::Message as _;
if chunk.snapshot_id.is_empty() {
let run = self
.run
.get_or_insert_with(|| LogRun::new(chunk.seq, chunk.restore_head.max(chunk.seq)));
let took = run
.take(
chunk.seq,
chunk.blob_len,
chunk.blob.len() as u64,
chunk.last,
)
.map(|()| run.through());
match took {
Err(why) => return Err(self.broke(why).await),
Ok(through) if chunk.last => {
self.through = through;
self.applied += 1;
}
Ok(_) => {}
}
}
self.tx
.send(Ok(frame(&chunk.encode_to_vec()).into()))
.await
.map_err(|_| gunnar_repair::Fault::new("cancelled", "the operator stopped reading"))
}
async fn close(self) -> Result<gunnar_repair::Restored, gunnar_repair::Fault> {
if let Some(Err(why)) = self.run.as_ref().map(LogRun::ended) {
return Err(self.broke(why).await);
}
self.tx
.send(Ok(bytes::Bytes::from_static(&END)))
.await
.map_err(|_| gunnar_repair::Fault::new("cancelled", "the operator stopped reading"))?;
Ok(gunnar_repair::Restored {
head_seq: self.through,
applied: self.applied,
lost: 0,
restart_required: false,
detail: "the copy ended whole".into(),
})
}
async fn abort(self, why: &str) {
let _ = self
.tx
.send(Err(std::io::Error::other(why.to_owned())))
.await;
}
}
async fn get_backup(State(c): State<Conn>) -> Response {
if let Some(r) = not_operator(&c, "the forge's history") {
return r;
}
let (_, target) = (c.ctx.now)();
let Some(target) = target else {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"backup-no-primary",
"no primary is known yet (the installer has not fed this front)",
);
};
let key = match tokio::fs::read_to_string(c.ctx.root.join(crate::console::IDENTITY_REL))
.await
.map_err(|e| e.to_string())
.and_then(|pem| {
gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).map_err(|e| e.to_string())
}) {
Ok(k) => k,
Err(e) => {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"backup-no-console-key",
format!("the console's key cannot be read: {e}"),
)
}
};
let mirror_at = |at: &str| -> Result<
gunnar_proto::mirror_client::MirrorClient<gunnar_grpc::client_signer::ControlChannel>,
String,
> {
Ok(
gunnar_proto::mirror_client::MirrorClient::new(mirror_channel(at)?)
.max_decoding_message_size(64 << 20),
)
};
// The TWIN when it is following and its log reads back ([`twin_verdict`]); else the PRIMARY
// (a twin mid-growth has its set unmounted: MEASURED release-3 mock 2026-09-29).
let mut chosen: Option<(&'static str, String, String)> = None;
if let Some(t) = &target.twin {
let verdict = match mirror_at(t) {
Err(e) => Err(e),
Ok(mut m) => {
let head = tokio::time::timeout(
FETCH_WAIT,
m.log_head(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::LogHeadRequest::default(),
)),
)
.await;
let head = match head {
Ok(Ok(h)) => Ok(h.into_inner()),
Ok(Err(s)) => Err(s.message().to_owned()),
Err(_) => Err(format!("no answer within {} s", FETCH_WAIT.as_secs())),
};
let probe = match &head {
Ok(h) if h.head_seq >= h.first_seq.max(1) => {
let from = h.first_seq.max(1);
let r = tokio::time::timeout(FETCH_WAIT, async {
let mut s = m
.log_fetch(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::LogFetchRequest {
from,
to: from,
max_bytes: 1,
from_offset: 0,
},
))
.await
.map_err(|s| s.message().to_owned())?
.into_inner();
s.message()
.await
.map(|_| ())
.map_err(|s| s.message().to_owned())
})
.await;
r.unwrap_or_else(|_| Err("no answer".into()))
}
_ => Ok(()),
};
twin_verdict(
head.as_ref().map_err(String::as_str),
probe.as_ref().map(|_| ()).map_err(String::as_str),
)
}
};
match verdict {
Ok(()) => chosen = Some(("twin", t.clone(), String::new())),
Err(why) => {
tracing::warn!(twin = %t, %why, "plugin door: the off-site copy is read from the PRIMARY instead of the twin");
chosen = Some(("primary", target.control.clone(), why));
}
}
}
let (source, at, source_why) =
chosen.unwrap_or(("primary", target.control.clone(), "no twin is fed".into()));
let mut mirror = match mirror_at(&at) {
Ok(m) => m,
Err(e) => return refused(StatusCode::INTERNAL_SERVER_ERROR, "backup-address", e),
};
let head = match tokio::time::timeout(
FETCH_WAIT,
mirror.log_head(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::LogHeadRequest::default(),
)),
)
.await
{
Ok(Ok(h)) => {
let h = h.into_inner();
// The chosen source must read back BEFORE a byte is streamed: a set held for its
// own growth (unmounted) is a NAMED refusal here, never a stream that breaks.
if source == "primary" && h.head_seq >= h.first_seq.max(1) {
let from = h.first_seq.max(1);
let probe = tokio::time::timeout(FETCH_WAIT, async {
let mut s = mirror
.log_fetch(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::LogFetchRequest {
from,
to: from,
max_bytes: 1,
from_offset: 0,
},
))
.await
.map_err(|s| s.message().to_owned())?
.into_inner();
s.message()
.await
.map(|_| ())
.map_err(|s| s.message().to_owned())
})
.await
.unwrap_or_else(|_| Err("no answer".into()));
if let Err((clause, line)) = primary_verdict(
&source_why,
&at,
probe.as_ref().map(|_| ()).map_err(String::as_str),
) {
return refused(StatusCode::SERVICE_UNAVAILABLE, clause, line);
}
}
h
}
Ok(Err(s)) => {
return refused(
StatusCode::BAD_GATEWAY,
"backup-source",
format!(
"the {source} at {at} answered Mirror.LogHead: {}",
s.message()
),
)
}
Err(_) => {
return refused(
StatusCode::GATEWAY_TIMEOUT,
"backup-source",
format!(
"the {source} at {at} did not answer Mirror.LogHead within {} s",
FETCH_WAIT.as_secs()
),
)
}
};
let (first, last, bytes, log_id) = (
head.first_seq,
head.head_seq,
head.bytes,
head.log_id.clone(),
);
tracing::info!(source, %at, first, last, bytes, "plugin door: the operator takes an off-site copy of the forge");
let (tx, rx) = tokio::sync::mpsc::channel::<Result<bytes::Bytes, std::io::Error>>(16);
// The ONE pump (gunnar-repair): the copy is a fresh box's `Bootstrap.Restore` written to
// frames instead of a socket, so it is exactly what a recharge would stream.
let channel = match mirror_channel(&at) {
Ok(c) => c,
Err(e) => return refused(StatusCode::INTERNAL_SERVER_ERROR, "backup-address", e),
};
let source_box = Watched(gunnar_repair::proto::ProtoTwin {
addr: at.clone(),
channel,
key,
});
let opts = copy_opts(source);
// A32-14, stays a task: the copy of one operator stream; the stream's receiver ends it.
tokio::spawn(async move {
match gunnar_repair::recharge_from_twin(&source_box, &FrameCopy(tx.clone()), &opts, &|| {
false
})
.await
{
Ok(r) if r.phase == gunnar_repair::Phase::Done => tracing::info!(
first,
last,
"plugin door: the off-site copy's stream ended whole"
),
Ok(r) => {
tracing::error!(why = %r.why, "plugin door: the off-site copy's stream BROKE — it ends without its end marker, so it is not kept")
}
Err(refusal) => {
tracing::error!(why = %refusal, "plugin door: the off-site copy's stream BROKE — it ends without its end marker, so it is not kept");
let _ = tx
.send(Err(std::io::Error::other(refusal.to_string())))
.await;
}
}
});
let mut r = Response::new(Body::from_stream(
tokio_stream::wrappers::ReceiverStream::new(rx),
));
let h = r.headers_mut();
h.insert(
"content-type",
HeaderValue::from_static("application/x-gunnar-log-frames"),
);
for (k, v) in [
("x-gunnar-source", source.to_string()),
("x-gunnar-source-why", source_why),
("x-gunnar-log-id", log_id),
("x-gunnar-first-seq", first.to_string()),
("x-gunnar-head-seq", last.to_string()),
("x-gunnar-bytes", bytes.to_string()),
] {
if let Ok(v) = HeaderValue::from_str(&v) {
h.insert(k, v);
}
}
r
}
/// How long an off-site copy waits for the source's next message.
pub(crate) const MESSAGE_WAIT: std::time::Duration = std::time::Duration::from_secs(120);
/// **One message of an off-site copy's stream, or a named refusal** once
/// `wait` has passed without one (stinky row 17, 2026-10-10): a source that
/// went silent mid-stream held the copy — and the operator's download — open
/// for ever.
pub(crate) async fn within<T>(
wait: std::time::Duration,
what: &str,
next: impl std::future::Future<Output = Result<T, String>>,
) -> Result<T, String> {
tokio::time::timeout(wait, next).await.map_err(|_| {
format!(
"{what}: nothing within {} s — the source went silent, so the copy is given up on",
wait.as_secs()
)
})?
}
/// **Where an off-site copy's log stream stands**: the last entry it holds
/// whole, of the run `from..=to` it asked for. Each entry must follow the last
/// one held (no hole), stay inside the run, and end with exactly its
/// `blob_len` bytes (stinky row 17, 2026-10-10) — the copy wrote every chunk
/// it was handed, so a stream with a hole still ended with its END marker.
pub(crate) struct LogRun {
through: u64,
to: u64,
/// The entry being put together, and the bytes of it seen so far.
open: Option<(u64, u64)>,
}
impl LogRun {
/// A run that asked for `from..=to`.
pub(crate) fn new(from: u64, to: u64) -> LogRun {
LogRun {
through: from.saturating_sub(1),
to,
open: None,
}
}
/// Take one chunk: entry `seq`, a blob of `blob_len` bytes in all, `len`
/// of them in this chunk, `last` when the entry ends with it.
pub(crate) fn take(
&mut self,
seq: u64,
blob_len: u64,
len: u64,
last: bool,
) -> Result<(), String> {
let got = match self.open {
Some((open, got)) if open == seq => got,
Some((open, got)) => {
return Err(format!(
"the log stream moved to entry {seq} with entry {open} cut at {got} bytes"
))
}
None => {
let want = self.through + 1;
if seq != want {
return Err(format!("the log stream sent entry {seq} where {want} was next — a copy with a hole in it is not kept"));
}
if seq > self.to {
return Err(format!(
"the log stream sent entry {seq}, past the {} asked for",
self.to
));
}
0
}
};
let got = got.saturating_add(len);
if got > blob_len {
return Err(format!(
"entry {seq} of the log stream runs past its {blob_len} bytes"
));
}
if last {
if got != blob_len {
return Err(format!(
"entry {seq} of the log stream ended at {got} of its {blob_len} bytes"
));
}
self.through = seq;
self.open = None;
} else {
self.open = Some((seq, got));
}
Ok(())
}
/// The stream ended: at an entry boundary, as the source promises.
pub(crate) fn ended(&self) -> Result<(), String> {
match self.open {
Some((seq, got)) => Err(format!(
"the log stream ended inside entry {seq}, {got} bytes in"
)),
None => Ok(()),
}
}
/// The last entry held whole.
pub(crate) fn through(&self) -> u64 {
self.through
}
}
/// `POST /feed/<key>` body.
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
pub struct Rotate {
pub value: String,
}
/// Pure: may `key` be rotated to `value`? A refusal names the key, never the value.
pub fn rotation_refusal(key: &str, value: &str) -> Option<(String, String)> {
if !ROTATABLE.contains(&key) {
return Some((
"feed-not-rotatable".into(),
format!(
"`{key}` is not one of the credentials this door rotates ({})",
ROTATABLE.join(", ")
),
));
}
let v = value.trim();
if v.is_empty() {
return Some((
"feed-rotate-empty".into(),
format!("an empty `{key}` would switch the credential off, not rotate it"),
));
}
if key == crate::console::FEED_ADMIN_TOKEN {
let mut feed = std::collections::BTreeMap::new();
feed.insert(key.to_string(), v.to_string());
if let Err(e) = crate::console::admin_token(&feed) {
return Some((
"console-admin-token-weak".into(),
e.trim_start_matches("REFUSED [console-admin-token-weak] ")
.to_string(),
));
}
}
None
}
/// **The primary's `Sealed` client, signed with this front's certify key** —
/// the machine principal `front-certify`, which owns the slots this front stores.
pub(crate) async fn sealed_client(
root: &std::path::Path,
control: &str,
) -> Result<
(
gunnar_core::ssh_key::PrivateKey,
gunnar_proto::sealed_client::SealedClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
gunnar_grpc::client_signer::CredentialSigner,
>,
>,
),
String,
> {
let key = crate::keys::certify_key(root).map_err(|e| format!("the certify key: {e:#}"))?;
let channel = tonic::transport::Channel::from_shared(format!("http://{control}"))
.map_err(|e| format!("{control}: {e}"))?
.connect_timeout(FETCH_WAIT)
.connect_lazy();
Ok((
key,
gunnar_proto::sealed_client::SealedClient::new(
tonic::service::interceptor::InterceptedService::new(
channel,
gunnar_grpc::client_signer::CredentialSigner,
),
),
))
}
/// **Store `value` in this front's slot `name` on the primary at `control`.**
pub async fn store_slot(
root: &std::path::Path,
control: &str,
name: &str,
value: Vec<u8>,
) -> Result<bool, String> {
let (key, mut c) = sealed_client(root, control).await?;
let req = gunnar_proto::SealedPutRequest {
name: name.to_string(),
value,
owner: String::new(),
};
match tokio::time::timeout(
FETCH_WAIT,
c.put(gunnar_grpc::client_signer::authed(&key, req)),
)
.await
{
Ok(Ok(r)) => Ok(r.into_inner().created),
Ok(Err(s)) => Err(format!(
"the primary at {control} answered Sealed.Put: {}",
s.message()
)),
Err(_) => Err(format!(
"the primary at {control} did not answer Sealed.Put within {} s",
FETCH_WAIT.as_secs()
)),
}
}
/// **Read this front's slot `name` from the primary at `control`.**
pub async fn read_slot(
root: &std::path::Path,
control: &str,
name: &str,
) -> Result<Vec<u8>, String> {
let (key, mut c) = sealed_client(root, control).await?;
match tokio::time::timeout(
FETCH_WAIT,
c.get(gunnar_grpc::client_signer::authed(
&key,
gunnar_proto::SealedGetRequest {
name: name.to_string(),
},
)),
)
.await
{
Ok(Ok(r)) => Ok(r.into_inner().value),
Ok(Err(s)) => Err(format!(
"the primary at {control} answered Sealed.Get: {}",
s.message()
)),
Err(_) => Err(format!(
"the primary at {control} did not answer Sealed.Get within {} s",
FETCH_WAIT.as_secs()
)),
}
}
/// `PUT /slot/<name>` — the operator's only.
async fn put_slot(
State(c): State<Conn>,
UrlPath(name): UrlPath<String>,
body: axum::body::Bytes,
) -> Response {
if let Some(r) = not_operator(&c, "storing a sealed slot") {
return r;
}
if body.is_empty() {
return refused(
StatusCode::BAD_REQUEST,
"slot-empty",
"an empty body stores nothing",
);
}
let (_, target) = (c.ctx.now)();
let Some(target) = target else {
return refused(
StatusCode::SERVICE_UNAVAILABLE,
"slot-no-primary",
"no primary is known yet (the installer has not fed this front)",
);
};
match store_slot(&c.ctx.root, &target.control, &name, body.to_vec()).await {
Ok(created) => {
// The installer material is held in this process too, while the primary lives:
// a lost primary's slot is read from here (the recharge), never from this disk.
if name == INSTALLER_MATERIAL_SLOT {
crate::recharge::hold(&body);
}
tracing::warn!(slot = %name, created, bytes = body.len(), "plugin door: the operator stored a sealed slot on the primary (the value is not logged)");
(StatusCode::OK, Json(serde_json::json!({"outcome": "STORED", "created": created, "line": format!("STORED `{name}` ({} bytes) in the primary's sealed slot, owned by this front's certify key", body.len())}))).into_response()
}
Err(e) => refused(StatusCode::BAD_GATEWAY, "slot-primary", e),
}
}
async fn post_feed(
State(c): State<Conn>,
UrlPath(key): UrlPath<String>,
Json(req): Json<Rotate>,
) -> Response {
if let Some(r) = not_operator(&c, "rotating a credential") {
return r;
}
if let Some((clause, why)) = rotation_refusal(&key, &req.value) {
return refused(StatusCode::BAD_REQUEST, &clause, why);
}
let rotate = std::sync::Arc::clone(&c.ctx.rotate);
let value = req.value.trim().to_string();
let k = key.clone();
match tokio::task::spawn_blocking(move || rotate(&k, &value)).await {
Ok(Ok(changed)) => {
let word = if changed { "ROTATED" } else { "UNCHANGED" };
tracing::warn!(key = %key, outcome = word, "plugin door: the operator rotated a fed credential (the value is not logged)");
(StatusCode::OK, Json(serde_json::json!({"outcome": word, "line": format!("{word} `{key}`: the console takes it on its next look (2 s); the value is kept sealed on this front")}))).into_response()
}
Ok(Err(e)) => refused(
StatusCode::INTERNAL_SERVER_ERROR,
"feed-rotate-unwritable",
e,
),
Err(e) => refused(
StatusCode::INTERNAL_SERVER_ERROR,
"feed-rotate-unwritable",
e.to_string(),
),
}
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ Main rebuild: the look answers the main as the guard reads it and the twin's review.
#[test]
fn the_rebuild_look_is_the_main_as_the_guard_reads_it_and_the_twins_review() {
let silent = rebuild_look_json("10.6.0.10:50051", Err("no answer within 10 s"), None);
assert_eq!(silent["main"]["seen"], "silent");
assert!(silent["main"]["head"].is_null() && silent["twin"].is_null());
assert_eq!(silent["main"]["why"], "no answer within 10 s");
let fresh = gunnar_repair::MainStatus {
window_open: true,
..Default::default()
};
let review = gunnar_repair::TwinReview {
twin_addr: "10.6.0.12:50051".into(),
recharge_possible: true,
head_seq: 77,
..Default::default()
};
let j = rebuild_look_json("10.6.0.10:50051", Ok(&fresh), Some(&review));
assert_eq!(
j["main"]["seen"],
serde_json::json!({"says": {"data": "fresh"}})
);
assert_eq!(j["main"]["head"], 0);
// The kit reads both back into the guard's own types.
let m: gunnar_repair::rebuild::Main =
serde_json::from_value(j["main"]["seen"].clone()).unwrap();
let t: gunnar_repair::TwinReview = serde_json::from_value(j["twin"].clone()).unwrap();
assert_eq!(
(m, t.head_seq, t.recharge_possible),
(
gunnar_repair::rebuild::Main::Says {
data: "fresh".into()
},
77,
true
)
);
}
/// Stinky row 17, its second half: a main that stops talking mid-stream
/// held the operator's copy open for ever.
#[tokio::test]
async fn a_source_that_goes_silent_mid_stream_is_given_up_on() {
let silent = std::future::pending::<Result<u8, String>>();
let read = tokio::time::timeout(
std::time::Duration::from_secs(10),
within(
std::time::Duration::from_millis(100),
"the log stream after 6",
silent,
),
)
.await;
let read = read.expect("the wait is the copy's, not the source's");
let why = read.unwrap_err();
assert!(why.contains("the log stream after 6"), "{why}");
assert_eq!(
within(std::time::Duration::from_secs(1), "x", async {
Ok::<_, String>(7u8)
})
.await,
Ok(7)
);
}
/// Stinky row 17 (2026-10-10): the off-site copy wrote every chunk it was
/// handed, so a stream that skipped an entry still ended with its END
/// marker — a copy with a hole in it, kept as whole.
#[test]
fn an_off_site_copy_refuses_a_stream_with_a_hole_in_it() {
let mut run = LogRun::new(5, 9);
run.take(5, 3, 3, true).unwrap();
run.take(6, 10, 4, false).unwrap();
run.take(6, 10, 6, true).unwrap();
assert_eq!(run.through(), 6);
assert!(run.take(8, 1, 1, true).is_err(), "7 was skipped");
let mut run = LogRun::new(5, 9);
assert!(run.take(6, 1, 1, true).is_err(), "the run starts at 5");
let mut run = LogRun::new(5, 9);
run.take(5, 10, 4, false).unwrap();
assert!(run.take(6, 1, 1, true).is_err(), "5 never ended");
let mut run = LogRun::new(5, 9);
assert!(run.take(5, 10, 4, true).is_err(), "5 ended 6 bytes short");
let mut run = LogRun::new(5, 5);
run.take(5, 1, 1, true).unwrap();
assert!(run.take(6, 1, 1, true).is_err(), "past the run asked for");
}
#[test]
fn the_twin_health_reading_is_the_heard_twin_furthest_behind_with_its_own_load() {
use gunnar_proto::{HealthResponse, TwinHeartbeat};
let row = |p: &str, state: &str, lag: u64, busy: Option<u64>| TwinHeartbeat {
principal: p.into(),
state: state.into(),
lag_entries: lag,
cpu_millicores_busy: busy,
cpu_millicores_total: busy.map(|_| 2_000),
..Default::default()
};
let h = HealthResponse {
twins: vec![
row("t1", "ok", 3, Some(400)),
row("t2", "behind", 25_000, Some(1_900)),
row("t3", "silent", 90_000, None),
],
..Default::default()
};
let v = twin_health_json(&h);
assert_eq!(v["twins"].as_array().unwrap().len(), 3);
assert_eq!(
(
v["reading"]["principal"].as_str(),
v["reading"]["lag_entries"].as_u64(),
v["reading"]["cpu_millicores_busy"].as_u64(),
v["reading"]["cpu_millicores_total"].as_u64()
),
(Some("t2"), Some(25_000), Some(1_900), Some(2_000)),
"a silent twin's old lag is not a reading: {v}"
);
let silent = twin_health_json(&HealthResponse {
twins: vec![row("t3", "silent", 90_000, None)],
..Default::default()
});
assert_eq!(silent["reading"]["state"], "silent");
assert_eq!(
twin_health_json(&HealthResponse::default())["reading"]["state"],
"none"
);
}
#[test]
fn only_the_fed_credentials_rotate_and_a_weak_bearer_is_refused() {
assert_eq!(
rotation_refusal("tls.gunnar.rs.key", "x").unwrap().0,
"feed-not-rotatable"
);
assert_eq!(
rotation_refusal(crate::console::FEED_ADMIN_TOKEN, " ")
.unwrap()
.0,
"feed-rotate-empty"
);
assert_eq!(
rotation_refusal(crate::console::FEED_ADMIN_TOKEN, "hunter2")
.unwrap()
.0,
"console-admin-token-weak"
);
assert!(rotation_refusal(
crate::console::FEED_ADMIN_TOKEN,
"qWJ8Zr3kP0sT7uV1xY4bN6mC9dF2gH5jL8aQ1wE4rT7yU0iO3pS6dF9gH2jK5lZ8"
)
.is_none());
let (_, why) = rotation_refusal(crate::console::FEED_ADMIN_TOKEN, "").unwrap();
assert!(!why.contains("hunter2"), "a refusal never carries a value");
}
/// ONE monetize (owner 13:35Z 2026-10-10): the front holds no Stripe or UpCloud secret,
/// so the door rotates none — the console's bearer is the one fed credential.
#[test]
fn the_front_rotates_no_monetize_secret() {
for k in [
"secret.stripe_key",
"secret.stripe_webhook",
"secret.upcloud_token",
] {
assert_eq!(
rotation_refusal(k, "sk_test_new").map(|r| r.0).as_deref(),
Some("feed-not-rotatable"),
"{k}"
);
}
assert_eq!(ROTATABLE, [crate::console::FEED_ADMIN_TOKEN]);
}
#[test]
fn a_twin_mid_growth_or_unreadable_is_not_the_source() {
let h = |state: &str| gunnar_proto::LogHeadResponse {
twin: true,
first_seq: 1,
head_seq: 694,
state: state.into(),
..Default::default()
};
assert_eq!(twin_verdict(Ok(&h("following")), Ok(())), Ok(()));
assert_eq!(
twin_verdict(Ok(&h("idle")), Ok(())),
Ok(()),
"not polled since start: its log is on disk"
);
let e = twin_verdict(Ok(&h("maintenance")), Ok(())).unwrap_err();
assert!(e.contains("maintenance"), "{e}");
for bad in ["snapshot-needed", "stalled", "diverged"] {
assert!(twin_verdict(Ok(&h(bad)), Ok(())).is_err(), "{bad}");
}
let e = twin_verdict(
Ok(&h("following")),
Err("txlog: No such file or directory (os error 2)"),
)
.unwrap_err();
assert!(e.contains("No such file"), "the measured failure: {e}");
assert!(twin_verdict(Err("unreachable"), Ok(())).is_err());
}
#[test]
fn a_primary_that_does_not_read_back_is_a_named_refusal_before_a_byte() {
// Defect B (release-3 mock 2026-09-29): with the twin passed over, the primary is the
// source; when its set is held for its own growth (unmounted) the door must refuse BY
// NAME before streaming, naming both reasons — never a copy that breaks mid-way.
assert_eq!(
primary_verdict(
"the twin is `maintenance` (growing), not following",
"10.0.0.1:50051",
Ok(())
),
Ok(())
);
let (clause, line) = primary_verdict(
"the twin is `maintenance` (growing), not following",
"10.0.0.1:50051",
Err("txlog: No such file or directory (os error 2)"),
)
.unwrap_err();
assert_eq!(clause, "backup-source-unreadable");
assert!(
line.contains("`maintenance` (growing)")
&& line.contains("10.0.0.1:50051")
&& line.contains("No such file"),
"both reasons: {line}"
);
assert!(line.contains("Nothing was streamed"), "{line}");
let (_, line) = primary_verdict("", "p:1", Err("no answer")).unwrap_err();
assert!(
line.contains("the twin: not fed"),
"no twin in the feed: {line}"
);
}
#[test]
fn a_chunk_is_framed_and_the_copy_asks_the_one_pump_for_a_fresh_box() {
assert_eq!(frame(b"abc"), vec![0, 0, 0, 3, b'a', b'b', b'c']);
// The pump plans the copy as a recharge into a fresh box: from the base when the
// source's log starts after 1, every chunk stamped with head, size and log id.
let head = gunnar_repair::TwinHead {
twin: true,
first_seq: 5,
head_seq: 9,
bytes: 1234,
log_id: "L".into(),
snapshot_at: 4,
snapshot_bytes: 10,
state: "following".into(),
..Default::default()
};
let plan = gunnar_repair::plan_recharge(
"t:1",
&fresh_box(),
&head,
gunnar_repair::TwinBase::Resumable,
©_opts("twin"),
)
.expect("a fresh box takes the twin");
assert_eq!(
(
plan.from,
plan.head,
plan.lost,
plan.announced_bytes,
plan.log_id.as_str()
),
(5, 9, 0, Some(1234), "L")
);
assert!(plan.base.is_some(), "the base streams first");
let primary = gunnar_repair::TwinHead {
twin: false,
..head
};
assert!(gunnar_repair::source_is_a_twin("p:1", &primary, ©_opts("twin")).is_err());
assert!(
gunnar_repair::source_is_a_twin("p:1", &primary, ©_opts("primary")).is_ok(),
"the door may copy the primary"
);
}
/// **The off-site copy is the ONE pump, over the real wire**: a primary's control
/// plane on loopback, copied as the door copies it — every chunk a frame stamped with
/// the head, the size and the log id, the last frame [`END`]; and an operator who stops
/// reading leaves a stream with no end marker (never a copy that looks whole).
#[test]
fn the_off_site_copy_streams_the_one_pump_into_frames_and_ends_whole() {
use prost::Message as _;
use std::sync::Arc;
let d = tempfile::tempdir().unwrap();
let stores = Arc::new(gunnar_server::stores::ServerStores::open(d.path()).unwrap());
for ns in ["a", "b"] {
stores.catalog().create_namespace(ns).unwrap();
stores
.create_repo(
gunnar_core::NewRepo::new(gunnar_core::RepoRef::new(ns, "app").unwrap()),
false,
)
.unwrap();
}
let head = stores.txlog().head_seq();
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(gunnar_server::CoreControlPlane::new(Arc::clone(&stores)));
let ctx = gunnar_grpc::ServiceContext::new(
plane,
Arc::new(gunnar_grpc::EventBus::new(16)),
gunnar_grpc::ServerConfig::default().insecure_allow_all_for_local_development(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let addr = std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap();
rt.spawn(async move {
let _ = gunnar_grpc::router(ctx, || 0).serve(addr).await;
});
let t0 = std::time::Instant::now();
while std::net::TcpStream::connect(addr).is_err() {
assert!(
t0.elapsed() < Duration::from_secs(10),
"the primary listens"
);
std::thread::sleep(Duration::from_millis(25));
}
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(
std::str::from_utf8(&gunnar_core::mint_ed25519().unwrap()).unwrap(),
)
.unwrap();
let at = addr.to_string();
let source = gunnar_repair::proto::ProtoTwin {
addr: at.clone(),
channel: rt.block_on(async { mirror_channel(&at) }).unwrap(),
key,
};
let (tx, mut rx) = tokio::sync::mpsc::channel(4096);
let r = rt
.block_on(gunnar_repair::recharge_from_twin(
&source,
&FrameCopy(tx),
©_opts("primary"),
&|| false,
))
.expect("the door may copy a primary");
assert_eq!(
(r.phase, r.head_seq),
(gunnar_repair::Phase::Done, head),
"{r:?}"
);
let mut frames = Vec::new();
while let Ok(f) = rx.try_recv() {
frames.push(f.expect("a whole copy carries no error"));
}
assert_eq!(
frames.last().map(|f| f.as_ref()),
Some(&END[..]),
"the end marker last"
);
let chunks: Vec<gunnar_proto::LogChunk> = frames[..frames.len() - 1]
.iter()
.map(|f| gunnar_proto::LogChunk::decode(&f[4..]).unwrap())
.collect();
assert!(
chunks
.iter()
.all(|c| c.restore_head == head && !c.log_id.is_empty()),
"every chunk stamped"
);
assert_eq!(
chunks.iter().filter(|c| c.last).map(|c| c.seq).max(),
Some(head)
);
assert_eq!(chunks.first().map(|c| c.seq), Some(1));
// The operator hangs up after the first frame: the copy breaks, no END is ever sent.
let (tx, mut rx) = tokio::sync::mpsc::channel(1);
let copy = FrameCopy(tx);
let r = rt.block_on(async {
let first = tokio::spawn(async move { rx.recv().await.map(|f| f.is_ok()) });
let r =
gunnar_repair::recharge_from_twin(&source, ©, ©_opts("primary"), &|| false)
.await;
(r, first.await.unwrap())
});
let (r, first) = r;
assert_eq!(first, Some(true), "one frame was read");
assert_eq!(
r.expect("planned").phase,
gunnar_repair::Phase::Failed,
"a copy nobody reads to the end is not whole"
);
}
/// **J (release-3 mocks 2026-09-29): an off-site copy that runs as asked is INFO.** The
/// operator taking a copy and the stream ending whole are the door working — at WARN they
/// counted towards the log gate's WARN budget and read as trouble on the console. A stream
/// that BREAKS stays an ERROR, and a fall-back to the primary stays a WARN.
#[test]
fn an_off_site_copy_that_runs_as_asked_is_said_at_info() {
let src = include_str!("operator_door.rs");
let level_of = |message: &str| -> String {
let at = src
.find(&format!("\"plugin door: {message}"))
.unwrap_or_else(|| panic!("no line says {message:?}"));
let call = src[..at].rfind("tracing::").expect("a tracing call");
src[call..at]
.split('!')
.next()
.unwrap()
.trim_start_matches("tracing::")
.to_string()
};
assert_eq!(
level_of("the operator takes an off-site copy of the forge"),
"info"
);
assert_eq!(level_of("the off-site copy's stream ended whole"), "info");
assert_eq!(level_of("the off-site copy's stream BROKE"), "error");
assert_eq!(
level_of("the off-site copy is read from the PRIMARY instead of the twin"),
"warn"
);
}
}