crates/gunnar-ui-server/src/grpc.rs
//! The client half: one channel to gunnar, one credential per call.
//!
//! Every gRPC call this process makes carries a fresh single-use `sshsig`,
//! minted by [`crate::signer`] in the interceptor. There is no session to reuse
//! and nothing to cache; that is gunnar's design, not an oversight here.
//!
//! Two ceilings shape everything below, and both are gunnar's:
//!
//! * **Every list RPC except `Refs.Log` ignores `page_size`** and answers with
//! the entire collection in one message. tonic's default 4 MiB decode limit is
//! then the only backstop, and it fails as an opaque "message length too
//! large". [`MAX_DECODE`] raises it, and the API layer slices for the browser.
//! * **`Refs.ReadObject` refuses anything over 64 MiB outright** and caps a
//! single response at 4 MiB, so `max_bytes` is mandatory rather than optional
//! for large blobs.
//!
//! The third is not a ceiling but a budget: gunnar's replay guard remembers 8192
//! `(fingerprint, nonce)` pairs across a five-minute window, shared by the whole
//! server. Proxying browser polls one-for-one would spend that budget on
//! answers that have not changed, so the read-heavy calls go through
//! [`Cached`].
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
use tonic::codegen::InterceptedService;
use tonic::service::Interceptor;
use tonic::transport::Channel;
use tonic::{Request, Status};
use crate::pb::access_client::AccessClient;
use crate::pb::accounts_client::AccountsClient;
use crate::pb::admin_client::AdminClient;
/// **The appliance's word about who may pay for what.** Metered only: an open
/// gunnar has no money in it and does not serve the service, so its client is
/// not generated there either (see `build.rs`).
#[cfg(feature = "metered")]
use crate::pb::billing_client::BillingClient;
/// **The tenant's own plan**, as the appliance holds it. Metered only: an open
/// gunnar is sold nothing and has no entitlement to report.
#[cfg(feature = "metered")]
use crate::pb::entitlement_client::EntitlementClient;
use crate::pb::bootstrap_client::BootstrapClient;
use crate::pb::events_client::EventsClient;
use crate::pb::forge_client::ForgeClient;
use crate::pb::mirror_client::MirrorClient;
use crate::pb::policy_client::PolicyClient;
use crate::pb::pull_requests_client::PullRequestsClient;
use crate::pb::refs_client::RefsClient;
use crate::pb::repos_client::ReposClient;
use crate::pb::usage_client::UsageClient;
use crate::signer::{Signer, CREDENTIAL_METADATA_KEY};
/// Decode ceiling for a response from gunnar.
///
/// Raised from tonic's 4 MiB default because gunnar's list RPCs are unbounded
/// and its `ReadObject` response can sit at its own 4 MiB cap with the oid,
/// kind and size fields pushing the encoded message past it.
pub const MAX_DECODE: usize = 8 * 1024 * 1024;
/// **How often the channel asks gunnar whether the connection is still
/// there** — an HTTP/2 PING, also while no call is open.
///
/// Without it a connection whose far end vanished IN SILENCE is never noticed
/// until the next call is written into it. MEASURED on the KVM mock C,
/// 2026-09-30 (row `recharge-main-front-cut-off`): a main restored from its
/// twin reboots with `reboot(RB_AUTOBOOT)` (gunnar-server `narrate::reboot_now`)
/// — no FIN, no RST. The console's event stream sat on the dead connection for
/// thirty minutes, and the next call (the functest's `POST /api/namespaces`)
/// was answered by the main's new kernel with RST: `HTTP 502 transport
/// error`. A PING into such a connection meets the same RST at once, the
/// connection is closed, the event stream re-subscribes and the next call
/// dials again. gunnar's listener is tonic's and enforces no ping policy.
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(10);
/// How long a PING may go unanswered before the connection is given up — the
/// far end that is gone without even an RST (a box powered off, a route lost).
///
/// Twenty seconds, not five (R5.1-G, 2026-10-07): a main whose control-plane
/// workers were all in slow synchronous calls answered no PING for seconds, and
/// five seconds killed the console's connection with every call on it (test2's
/// `KeepAliveTimedOut` stalls). A main that reboots still answers RST at once.
pub const KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
/// A gunnar-authenticated channel.
type Authed = InterceptedService<Channel, SshsigInterceptor>;
/// **The metadata key a person's identity-core session travels under**
/// (IDENTITY-HOST, A34-18): gunnar's `auth::SESSION_METADATA_KEY`, the opaque
/// token `IdentityService` minted. A call carries it OR a signed credential,
/// never both (gunnar refuses the pair).
pub const SESSION_METADATA_KEY: &str = "nornir-identity-session";
/// Signs every outgoing request with the console's key — or, for a client
/// acting AS a person ([`Gunnar::as_session`]), carries that person's
/// identity-core session instead.
#[derive(Clone)]
pub struct SshsigInterceptor {
signer: Arc<Signer>,
/// The person's core session token: then no signature is made.
session: Option<Arc<str>>,
}
impl Interceptor for SshsigInterceptor {
fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
if let Some(token) = &self.session {
let value = token
.parse()
.map_err(|_| Status::internal("the session token is not a valid header value"))?;
request.metadata_mut().insert(SESSION_METADATA_KEY, value);
return Ok(request);
}
// **Bound to the method it is about to be spent on** (GUN-001). tonic's
// generated clients put the service and method into the extensions
// before the interceptor runs, so this is the one place that holds the
// key and the method at once. A request without it is an error and
// not a fall-back to the method-blind v1 shape: a downgrade nobody
// would see is how the finding happened.
let method = request
.extensions()
.get::<tonic::GrpcMethod>()
.map(|m| {
// `gunnar.v1.Access` + `AddKey` -> `Access.AddKey`, the one
// spelling gunnar's `auth::METHODS` and its
// `method_name_from_path` both use.
let service = m.service().rsplit('.').next().unwrap_or(m.service());
format!("{service}.{}", m.method())
})
.ok_or_else(|| {
Status::internal(
"this call carries no tonic::GrpcMethod, so its credential cannot be bound \
to the method it is about to be spent on",
)
})?;
let header = self
.signer
.credential_for_now(&method)
.map_err(|e| Status::internal(format!("cannot sign the control-plane call: {e}")))?;
let value = header
.parse()
.map_err(|_| Status::internal("the signed credential is not a valid header value"))?;
request
.metadata_mut()
.insert(CREDENTIAL_METADATA_KEY, value);
Ok(request)
}
}
/// Handles to gunnar's control plane.
///
/// Cheap to clone: the channel multiplexes and the clients are thin.
#[derive(Clone)]
pub struct Gunnar {
channel: Channel,
interceptor: SshsigInterceptor,
}
impl std::fmt::Debug for Gunnar {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Gunnar").finish_non_exhaustive()
}
}
impl Gunnar {
/// Connect lazily to gunnar's control plane.
///
/// The connection is not established here: gunnar may still be starting, and
/// a console that refuses to boot because its server has not finished
/// booting is worse than one that reports the server as unreachable.
///
/// # Errors
///
/// Fails only if `endpoint` is not a usable URI.
pub fn connect_lazy(endpoint: &str, signer: Arc<Signer>) -> Result<Gunnar, String> {
Ok(Gunnar::over(Gunnar::lazy_channel(endpoint)?, signer))
}
/// **The channel alone**, lazily connected — what [`Gunnar::connect_lazy`]
/// signs over, for a caller that signs as MANY people over ONE connection
/// ([`Gunnar::over`]).
///
/// # Errors
///
/// Fails only if `endpoint` is not a usable URI.
pub fn lazy_channel(endpoint: &str) -> Result<Channel, String> {
// Plaintext on purpose: gunnar's gRPC listener applies no TLS at all, so
// this connection must stay on loopback or a private interface.
let uri = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
endpoint.to_owned()
} else {
format!("http://{endpoint}")
};
Ok(Channel::from_shared(uri.clone())
.map_err(|e| format!("{uri:?} is not a usable control-plane address: {e}"))?
.connect_timeout(Duration::from_secs(5))
.http2_keep_alive_interval(KEEPALIVE_EVERY)
.keep_alive_timeout(KEEPALIVE_TIMEOUT)
.keep_alive_while_idle(true)
.connect_lazy())
}
/// **A client over an existing channel, signed with `signer`.** tonic's
/// `Channel` is a handle: its clones share ONE HTTP/2 connection and
/// multiplex every call over it, so a console that signs as each person
/// (their session key) needs one connection to gunnar, not one per person
/// ([`crate::tenant::TenantKeys`]: one per login was a file descriptor on
/// vm1 per login, kept for ever — MEASURED on the KVM mock 2026-10-06).
pub fn over(channel: Channel, signer: Arc<Signer>) -> Gunnar {
Gunnar {
channel,
interceptor: SshsigInterceptor { signer, session: None },
}
}
/// **The same channel, acting AS the person whose identity-core session
/// `token` is** (IDENTITY-HOST): every call carries the token, gunnar's gate
/// asks the core who it is, and their grants decide. The console decides
/// nothing about them.
pub fn as_session(&self, token: &str) -> Gunnar {
Gunnar {
channel: self.channel.clone(),
interceptor: SshsigInterceptor { signer: Arc::clone(&self.interceptor.signer), session: Some(Arc::from(token)) },
}
}
/// **The identity core** (`nornir.identity.v1.IdentityService`), mounted by
/// gunnar outside its gate: the calls a person makes before they hold any
/// credential (login, redeem), and the ones that carry their session in the
/// request itself.
pub fn identity(&self) -> nornir_identity::grpc::pb::identity_service_client::IdentityServiceClient<Channel> {
nornir_identity::grpc::pb::identity_service_client::IdentityServiceClient::new(self.channel.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **The same channel, signed with another key** — the MCP connector's
/// agent key (`crate::mcp`), which gunnar enrolled on the person.
pub fn as_signer(&self, signer: Arc<Signer>) -> Gunnar {
Gunnar::over(self.channel.clone(), signer)
}
/// **Another box, signed with this console's own key** — the twin, whose
/// identity table is replayed from the primary's and so knows this key.
pub fn sibling(&self, endpoint: &str) -> Result<Gunnar, String> {
Gunnar::connect_lazy(endpoint, Arc::clone(&self.interceptor.signer))
}
/// The console's own signing key (a person's session carries no key of
/// theirs: their client signs nothing).
pub fn signer(&self) -> Arc<Signer> {
Arc::clone(&self.interceptor.signer)
}
/// The fingerprint gunnar knows this console by.
pub fn fingerprint(&self) -> &str {
self.interceptor.signer.fingerprint()
}
/// **Clans** (2026-10-05): the person's own verbs — make one, invite,
/// answer, give room, leave — signed as the person; gunnar decides who
/// the clan's admin is.
pub fn clans(&self) -> crate::pb::clans_client::ClansClient<Authed> {
crate::pb::clans_client::ClansClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// Repository and namespace catalogue.
pub fn repos(&self) -> ReposClient<Authed> {
ReposClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// References, history and raw objects.
pub fn refs(&self) -> RefsClient<Authed> {
RefsClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// The pre-receive gate's configuration: branch protection, ref-name
/// policy, and which gates a repository actually has configured.
///
/// Not a disabled panel. gunnar serves all six of these RPCs against a
/// policy document at `refs/gunnar/policy`; what was missing was anything
/// on this side that could reach them.
pub fn policy(&self) -> PolicyClient<Authed> {
PolicyClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// Principals, keys and grants.
pub fn access(&self) -> AccessClient<Authed> {
AccessClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **Tickets: gunnar vouching that this caller may act for a namespace.**
///
/// Its own client for the reason [`accounts`](Gunnar::accounts) is its own:
/// gunnar draws the line, and `Billing` is the one service on this control
/// plane whose answer is meant to be carried OFF the box and presented to
/// somebody else. What makes that safe is that gunnar checks the caller's
/// grant on the namespace before a ticket exists, against the key the call
/// was signed with — so the credential this client is built on decides what
/// comes back. Calling it with the console's own server-wide admin key
/// would mint a ticket for anything; [`crate::api::subscription`] therefore
/// calls it through the TENANT's channel and never through `App::gunnar`.
#[cfg(feature = "metered")]
pub fn billing(&self) -> BillingClient<Authed> {
BillingClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **The last fact monetize pushed about a namespace**: plan, state,
/// `paid_until`, caps.
///
/// Read-only in practice, and the console has no key to make it otherwise:
/// gunnar verifies every `Entitlement.Set` against MONETIZE's Ed25519 key
/// before it writes, and this process holds no such key. `Get` is
/// namespace-scoped since gunnar `ae6ac29`, so calling it through a
/// TENANT's channel answers that tenant's own plan and refuses everyone
/// else's — which is how `api::subscription::plan` is written.
#[cfg(feature = "metered")]
pub fn entitlement(&self) -> EntitlementClient<Authed> {
EntitlementClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// Health, statistics and storage.
pub fn admin(&self) -> AdminClient<Authed> {
AdminClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **Per-namespace meters.** What one namespace holds, counted by gunnar.
///
/// Separate from [`admin`](Gunnar::admin) although both report bytes, and
/// the difference is who may see them: `Admin.StorageInfo` is one row per
/// repository across the whole server and takes a server-wide grant, while
/// `Usage.Read` takes `Read` on the ONE namespace it names. That split is
/// gunnar's, and it is why the console reads a namespace's own numbers
/// through here rather than by filtering the server-wide answer — filtering
/// would mean the console had already been handed every tenant's figures.
pub fn usage(&self) -> UsageClient<Authed> {
UsageClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **The per-USER verbs: the account gate, and the hard purge.**
///
/// Its own client and not part of [`repos`](Gunnar::repos) because gunnar
/// draws the same line: `Repos` acts on a path, `Accounts` acts on a name
/// that is simultaneously an identity and a namespace. Every RPC on it is
/// Admin at SERVER scope — the console signs as one principal that holds
/// exactly that, so what decides whether a human may press these buttons is
/// the BFF's own role floor (`/api/accounts` is an admin surface), not
/// gunnar's grant.
pub fn accounts(&self) -> AccountsClient<Authed> {
AccountsClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **Company single sign-on** (2026-10-05): an account's identity
/// provider, SCIM, the CSV import — and the relay's verbs (Discover,
/// Begin, Finish, SecondFactor, Scim), which gunnar decides.
pub fn sso(&self) -> crate::pb::sso_client::SsoClient<Authed> {
crate::pb::sso_client::SsoClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **The owner's enterprise list** (2026-10-05): the account admin's
/// audit log, the operator's aggregates and look, the name filter. Signed
/// as whoever acts; gunnar decides who may.
pub fn enterprise(&self) -> crate::pb::enterprise_client::EnterpriseClient<Authed> {
crate::pb::enterprise_client::EnterpriseClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **Pull mirrors: where a repository on this box came from, and the job
/// that is fetching it.**
///
/// This client did not exist until 2026-09-08, and `tests/contract.rs`
/// asserted that it did not — on the strength of an argument that has since
/// stopped being true. `Mirror.Sync` fetches, and fetching used to mean
/// `gunnar-client`: `gix`, a `.git` directory and a worktree, none of which
/// exist on an appliance running as PID 1 over znippy, so a server-side
/// crate could not link it under the dependency budget.
/// `gunnar_wire::fetch_pack` is a fetch client with none of those, and
/// `Mirror.Sync` is now an enqueue on the same one-at-a-time job runner
/// `Repos.Gc` uses.
///
/// One client for five verbs the console drives — `Add`, `SetCredential`,
/// `Sync`, `Status`, `Remove` — and six replication verbs it does not.
///
/// **`SetCredential` is the only call this console makes that carries a
/// secret INTO gunnar.** It goes one way: no RPC on this service returns a
/// credential, `MirrorConfig` has no field that could, and the BFF forgets
/// its copy the moment the call returns. See [`crate::api::mirrors`].
pub fn mirror(&self) -> MirrorClient<Authed> {
MirrorClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// Pull requests.
pub fn pulls(&self) -> PullRequestsClient<Authed> {
PullRequestsClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// The review half of a pull request: comments and signed approvals.
///
/// Separate from [`pulls`](Gunnar::pulls) because the two have different
/// authorities — the record is gunnar's under a compare-and-swap, and the
/// review is a log of signatures gunnar only verifies. `PullRequests.Merge`
/// is where they meet, and it refuses without a verified approval of the
/// tip.
pub fn forge(&self) -> ForgeClient<Authed> {
ForgeClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// **One forge event, ready for `Forge.Append` from this client.**
///
/// * The console's own client (and an agent key's) SIGNS it with its key:
/// the event is that key's word, checked by gunnar against its key table.
/// * A client acting AS A PERSON ([`Gunnar::as_session`]) leaves it
/// UNSIGNED (GHUNT-19): the person's passkey session holds no key of theirs
/// and the console must not sign in their name. Gunnar attests it itself —
/// its own key, the actor named — when the session stepped up, and answers
/// `[stepup]` otherwise, which the console relays as `"code":"stepup"` and
/// the web answers with a fresh touch in place.
///
/// # Errors
///
/// Fails only if signing itself fails, which is a fault rather than a
/// condition, so it is rendered as `Internal`.
pub fn sign_forge_event(
&self,
event: &crate::forge::ForgeEvent,
) -> Result<crate::pb::SignedForgeEvent, Status> {
let (key_fingerprint, signature) = if self.interceptor.session.is_some() {
(String::new(), String::new())
} else {
self.interceptor
.signer
.sign_forge_event(event)
.map_err(|e| Status::internal(format!("cannot sign the review event: {e}")))?
};
Ok(crate::pb::SignedForgeEvent {
event: Some(crate::pb::ForgeEvent {
repo: event.repo.clone(),
pull: event.pull,
actor: event.actor.clone(),
at_unix_ms: event.at_unix_ms,
kind: Some(match &event.event {
crate::forge::Event::Commented {
body,
path,
line,
blob_oid,
} => crate::pb::forge_event::Kind::Commented(crate::pb::Commented {
body: body.clone(),
path: path.clone(),
line: *line,
blob_oid: blob_oid.clone(),
}),
crate::forge::Event::Approved { commit } => {
crate::pb::forge_event::Kind::Approved(crate::pb::Approved {
commit: commit.clone(),
})
}
crate::forge::Event::IssueOpened { title, body } => {
crate::pb::forge_event::Kind::IssueOpened(crate::pb::IssueOpened {
title: title.clone(),
body: body.clone(),
})
}
crate::forge::Event::IssueCommented { body } => {
crate::pb::forge_event::Kind::IssueCommented(crate::pb::IssueCommented {
body: body.clone(),
})
}
crate::forge::Event::IssueClosed { reason } => {
crate::pb::forge_event::Kind::IssueClosed(crate::pb::IssueClosed {
reason: reason.clone(),
})
}
}),
}),
key_fingerprint,
signature,
attested: None,
})
}
/// The event stream.
pub fn events(&self) -> EventsClient<Authed> {
EventsClient::with_interceptor(self.channel.clone(), self.interceptor.clone())
.max_decoding_message_size(MAX_DECODE)
}
/// A fresh method-bound credential from this console's key, for a call that
/// no generated client makes (the raw `Bootstrap.Restore` stream).
pub fn credential_for(&self, method: &str) -> Result<String, String> {
self.interceptor.signer.credential_for_now(method).map_err(|e| e.to_string())
}
/// Bootstrap status.
///
/// The only service gunnar mounts outside its authentication gate, and so
/// the only call that works before this console has an enrolled identity.
pub fn bootstrap(&self) -> BootstrapClient<Channel> {
BootstrapClient::new(self.channel.clone()).max_decoding_message_size(MAX_DECODE)
}
/// `Bootstrap` as a raw gRPC client on the same channel, for calls that
/// need their own codec. The restore needs one (`gunnar_repair::grpc::PoisonCodec`)
/// because its request body must be able to FAIL.
pub fn bootstrap_grpc(&self) -> tonic::client::Grpc<Channel> {
tonic::client::Grpc::new(self.channel.clone()).max_decoding_message_size(MAX_DECODE)
}
/// **`Access.WhoAmI` carrying SOMEONE ELSE's credential** — the twin door's
/// step-up (5.4): a person signs a `gunnar-sshsig-v2` credential bound to
/// `Access.WhoAmI` with their own key, and this gunnar (the twin) says whose
/// key it is and what that principal may do. Relayed once, never minted here.
pub async fn who_am_i_as(&self, credential: &str) -> Result<crate::pb::WhoAmIResponse, tonic::Status> {
let mut request = tonic::Request::new(crate::pb::WhoAmIRequest {});
let value = credential
.trim()
.parse()
.map_err(|_| tonic::Status::unauthenticated("the step-up credential is not a valid header value"))?;
request.metadata_mut().insert(crate::signer::CREDENTIAL_METADATA_KEY, value);
AccessClient::new(self.channel.clone())
.max_decoding_message_size(MAX_DECODE)
.who_am_i(request)
.await
.map(tonic::Response::into_inner)
}
}
/// A refusal from gunnar, carried to the browser without rewording.
///
/// The console renders `message` verbatim. A paraphrase would be a second
/// source of truth about what the server decided, and the operator would have
/// no way to tell which one to believe.
#[derive(Debug, Clone, serde::Serialize)]
pub struct Fail {
/// gunnar's gRPC status code, lowercased.
pub code: String,
/// gunnar's own words.
pub message: String,
}
/// **monetize's sentence for the CUSTOMER, riding beside the operator's.**
///
/// `monetize_server::services::purchase::CUSTOMER_WORDS_KEY`. When monetize
/// refuses an order it puts the operator's reason in the status message — the
/// one that names the machine (`gunnar cannot serve 100.0 GiB … /data has
/// 39.8 GiB free`) — and, on this metadata key, the sentence the person who
/// pressed the button is meant to read. Both come from one writer over there
/// (`refused_status`), so a refusal cannot carry one without the other.
///
/// A `-bin` key because the sentence is UTF-8 and an em dash is not ASCII.
pub const CUSTOMER_WORDS_KEY: &str = "monetize-customer-words-bin";
/// The customer's sentence riding on `status`, when monetize sent one that is
/// not blank. The one reader of [`CUSTOMER_WORDS_KEY`].
pub fn customer_words(status: &Status) -> Option<String> {
status
.metadata()
.get_bin(CUSTOMER_WORDS_KEY)
.and_then(|v| v.to_bytes().ok())
.and_then(|b| String::from_utf8(b.to_vec()).ok())
.filter(|s| !s.trim().is_empty())
}
impl From<Status> for Fail {
/// The status as the browser will see it.
///
/// **The customer's words win the body when the server sent them; the
/// operator's words are what the body carries when it did not.** That
/// second half is the one this function must never grow out of: a status
/// with no customer sentence on it is shown as the server said it, and the
/// console composes nothing in the gap. MEASURED 2026-09-20 on gunnar.rs —
/// the owner pressed Buy on a bundle the box could not hold and the page
/// answered with a sentence about maintenance that nobody was carrying out.
/// That sentence was a fallback that could not be wrong, and a fallback that
/// cannot be wrong is the defect, not the safety net.
///
/// The operator's reason is not lost: it goes to the journal here, once, on
/// the same line as the words that replaced it, so the two can be read
/// against each other.
fn from(status: Status) -> Fail {
// gunnar's step-up refusal (A34-10: an SSO write from a session with no
// fresh user-verified touch) is the console's own step-up code, so the
// browser answers it with a touch in place and sends the call once more.
let code = if status.code() == tonic::Code::PermissionDenied && status.message().starts_with("[stepup]") {
crate::auth::STEP_UP_CODE.to_owned()
} else {
format!("{:?}", status.code()).to_lowercase()
};
let operators = status.message().to_owned();
let customers = customer_words(&status);
match customers {
Some(words) => {
tracing::warn!(
code = %code,
reason = %operators,
shown = %words,
"monetize refused the order; the customer reads the words, the reason is here"
);
Fail {
code,
message: words,
}
}
None => Fail {
code,
message: operators,
},
}
}
}
impl Fail {
/// The HTTP status this refusal deserves.
///
/// gunnar refusing the console's own credential is not the operator's
/// mistake, so it surfaces as a gateway fault rather than as a 401 the
/// browser would try to solve by logging in again.
pub fn http_status(&self) -> http::StatusCode {
use http::StatusCode as S;
match self.code.as_str() {
"notfound" => S::NOT_FOUND,
// The console's OWN refusals on an import (`api::mirror_gate::code`):
// not a namespace of yours, or one that needs more room — both a
// "you may not, as things stand" — and too many at once.
"notyours" | "needsroom" => S::FORBIDDEN,
// A session whose user-verified touch is too old (`auth::STEP_UP_CODE`).
"stepup" => S::FORBIDDEN,
"busy" => S::TOO_MANY_REQUESTS,
"alreadyexists" => S::CONFLICT,
"invalidargument" | "failedprecondition" | "outofrange" => S::BAD_REQUEST,
"unimplemented" => S::NOT_IMPLEMENTED,
"unavailable" => S::SERVICE_UNAVAILABLE,
// gunnar answers ResourceExhausted for two unrelated limits: a single
// object or response that is too large to serve (a payload problem →
// 413), and the admission gate being full (a concurrency / rate
// problem → 429). They are told apart by gunnar's own words; the
// verbatim message reaches the browser either way, so a
// mis-classification still tells the operator what happened.
"resourceexhausted" => {
if self.is_object_too_large() {
S::PAYLOAD_TOO_LARGE
} else {
S::TOO_MANY_REQUESTS
}
}
"deadlineexceeded" => S::GATEWAY_TIMEOUT,
// An authorisation answer about one resource, not a verdict on the
// console's credential. gunnar answers this for a repository that
// does not exist as readily as for one the caller may not read, and
// it does not say which — so neither does the console.
"permissiondenied" => S::FORBIDDEN,
// **A clock-skew refusal is temporary, so it is a 503.** gunnar says
// `Unauthenticated` both for a console key it has never heard of
// and for a perfectly good key whose `issued_at_ms` is more than
// 300 s from its own clock. The first is a registration fault and
// stays a gateway fault below; the second clears by itself once the
// clocks agree, which is exactly what 503 means and what the
// browser's "Try again" is for.
//
// It is ALSO the status that survives the proxy. Behind Cloudflare an
// origin 502 has its body replaced by Cloudflare's own HTML error
// page — measured 2026-09-14, when this refusal reached an operator
// as a screenful of raw Cloudflare markup instead of gunnar's words.
// Both estate faults (see `is_infrastructure_fault`) are 503, not
// 502, so a proxy does not swap the answer for its own error page.
// Every OTHER `Unauthenticated` stays a 502 with its words intact —
// above all a forge rejecting the operator's own token, which is the
// operator's to fix and must say so, naming the forge.
"unauthenticated" if self.is_infrastructure_fault() => S::SERVICE_UNAVAILABLE,
_ => S::BAD_GATEWAY,
}
}
/// True when a `ResourceExhausted` refusal is about a single object or
/// response being too large (→ 413), as opposed to the admission gate being
/// full (→ 429).
///
/// gunnar's two size refusals both name an `object … bytes` over a byte
/// `ceiling`/`limit` (`control_plane.rs`), while the admission gate speaks of
/// being full and in-flight transfers (`repos.rs`). Matching gunnar's own
/// words keeps this a read of the server's verdict, not a second opinion; the
/// verbatim message is shown regardless, so a miss still informs.
fn is_object_too_large(&self) -> bool {
let m = self.message.to_ascii_lowercase();
m.contains("ceiling for reading an object")
|| m.contains("a grpc client can receive")
|| (m.contains("object ")
&& m.contains(" bytes")
&& (m.contains("ceiling") || m.contains("limit")))
}
/// True when gunnar refused the console's credential itself.
///
/// Only `Unauthenticated` counts. It means the console's key is not in
/// gunnar's table, or the signature did not verify, and no amount of
/// retrying will fix it.
///
/// `PermissionDenied` deliberately does not: the console holds a
/// server-wide admin grant, so in practice gunnar answers it for a
/// repository that is not there, and treating that as a credential fault
/// would send an operator to check a key that is fine.
pub fn is_console_identity_problem(&self) -> bool {
self.code == "unauthenticated" && !self.is_clock_skew()
}
/// True when gunnar refused a credential only because the two machines'
/// clocks disagree by more than its window — not because the key is wrong.
///
/// Matched on gunnar's own words (`the credential is N ms away from this
/// server's clock`), the same way [`Self::is_object_too_large`] reads
/// gunnar's words, because gunnar does not yet put a machine-readable
/// reason on an auth failure. When it does, this should read that instead;
/// the words are the fallback, not the design.
/// **A fault of the estate, not of the person looking at the page.** The
/// clocks disagree, or the console's own key is not accepted. Whoever is
/// logged in — an operator or a tenant who bought an account — can do
/// nothing about either, and the words for them are "down for maintenance".
/// The reason is written to the journal by [`crate::api::ApiError`] and is
/// never put in front of them. Owner's ruling, 2026-09-14.
/// **gunnar's main is being restored** (`LogFollows`, gunnar bddb36d): UNAVAILABLE with
/// its sentence "this box is being restored (or is a twin): …". The browser gets the
/// console's one word for it (`restoring`), never "transaction log" or "twin"; the
/// journal keeps gunnar's message (wiring tracer B 2026-10-08).
pub fn is_restoring(&self) -> bool {
self.code == "unavailable" && self.message.contains("this box is being restored")
}
pub fn is_infrastructure_fault(&self) -> bool {
self.is_clock_skew() || self.is_console_key_refused()
}
/// gunnar did not accept the CONSOLE's own key — its own words,
/// `the credential did not verify against any registered key`.
///
/// Deliberately narrower than [`Self::is_console_identity_problem`], which
/// is true for every `Unauthenticated`: that one also covers a forge
/// rejecting the operator's own token during a mirror import, and turning
/// THAT into "down for maintenance" would hide from the operator the one
/// thing they can fix. The first version of this rule did exactly that and
/// `mirroring::a_forge_that_rejects_the_token_registers_nothing` caught it.
pub fn is_console_key_refused(&self) -> bool {
self.code == "unauthenticated" && self.message.contains("did not verify against any registered key")
}
pub fn is_clock_skew(&self) -> bool {
self.code == "unauthenticated" && self.message.contains("away from this server's clock")
}
}
impl std::fmt::Display for Fail {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"Gunnar refused the call ({}): {}",
self.code, self.message
)
}
}
/// A value held for a short while so repeated reads do not each cost a
/// credential.
///
/// The window is deliberately shorter than any human notices and long enough to
/// absorb a page full of panels asking the same question at once.
pub struct Cached<T> {
ttl: Duration,
held: Mutex<Option<(Instant, T)>>,
}
impl<T: Clone> Cached<T> {
/// A cache holding values for `ttl`.
pub fn new(ttl: Duration) -> Cached<T> {
Cached {
ttl,
held: Mutex::new(None),
}
}
/// Return the held value, or produce and hold a fresh one.
///
/// The lock is held across `refresh`, so concurrent callers wait for one
/// answer rather than each spending a credential on their own.
///
/// # Errors
///
/// Propagates whatever `refresh` fails with. A failure is not cached: the
/// next caller tries again.
pub async fn get<E, F>(&self, refresh: F) -> Result<T, E>
where
F: std::future::Future<Output = Result<T, E>>,
{
let mut held = self.held.lock().await;
if let Some((at, value)) = held.as_ref() {
if at.elapsed() < self.ttl {
return Ok(value.clone());
}
}
let fresh = refresh.await?;
*held = Some((Instant::now(), fresh.clone()));
Ok(fresh)
}
/// Drop the held value, so the next read goes to gunnar.
///
/// Called after a mutation: a console that shows a stale list right after
/// the operator changed it reads as a failed write.
pub async fn invalidate(&self) {
*self.held.lock().await = None;
}
/// [`Cached::get`] when the answer may be shared, the call itself when it
/// may not.
///
/// `shareable` is [`crate::api::Reader::may_cache`]: the held value is the
/// console key's answer, so a reader signed as a PERSON (or the guest) must
/// neither read it nor fill it — gunnar answers them, every time.
pub async fn get_for<E, F>(&self, shareable: bool, refresh: F) -> Result<T, E>
where
F: std::future::Future<Output = Result<T, E>>,
{
if shareable {
self.get(refresh).await
} else {
refresh.await
}
}
}
#[cfg(test)]
mod tests {
/// ★ **A34-10: gunnar's `[stepup]` refusal is the console's step-up
/// code.** Gunnar refuses an SSO write from a session with no fresh
/// user-verified touch; the browser must hear `"code":"stepup"` (403) so
/// it asks the person's key for a touch in place and sends the call once
/// more, not a plain "permission denied".
#[test]
fn gunnars_step_up_refusal_is_the_consoles_step_up_code() {
let fail = Fail::from(Status::permission_denied(
"[stepup] Sso.SetProvider needs a fresh touch of your own passkey (user-verified, within the last 10 minutes)",
));
assert_eq!(fail.code, crate::auth::STEP_UP_CODE);
assert_eq!(fail.http_status(), http::StatusCode::FORBIDDEN);
let plain = Fail::from(Status::permission_denied("not an admin of acme"));
assert_eq!(plain.code, "permissiondenied", "every other refusal stays what it was");
}
use super::*;
/// **Every outgoing call is signed for the method it calls.** The
/// interceptor reads `tonic::GrpcMethod`, which the generated clients put
/// in the extensions, and refuses a request without one rather than fall
/// back to the method-blind shape (GUN-001).
#[test]
fn the_interceptor_binds_the_credential_to_the_called_method() {
let seed = [9u8; 32];
let key = ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&seed));
let signer = Arc::new(Signer::new(key).expect("a usable signer"));
let mut interceptor = SshsigInterceptor {
signer: Arc::clone(&signer),
session: None,
};
let mut request = Request::new(());
request
.extensions_mut()
.insert(tonic::GrpcMethod::new("gunnar.v1.Repos", "CreateNamespace"));
let signed = interceptor.call(request).expect("signed");
let header = signed
.metadata()
.get(CREDENTIAL_METADATA_KEY)
.expect("the credential header")
.to_str()
.expect("ascii")
.to_owned();
let fields: Vec<&str> = header.split_whitespace().collect();
assert_eq!(fields[0], "gunnar-sshsig-v2", "{header}");
assert_eq!(fields[1], signer.fingerprint());
assert_eq!(
fields[4], "Repos.CreateNamespace",
"the package is dropped, the service kept"
);
assert!(crate::signer::verify_header(
&signer.public_key_openssh(),
&header
));
// No method, no credential — and an error, not a v1 header.
let bare = interceptor.call(Request::new(()));
let status = bare.expect_err("a request with no GrpcMethod is refused");
assert_eq!(status.code(), tonic::Code::Internal);
assert!(
status.message().contains("GrpcMethod"),
"{}",
status.message()
);
}
#[test]
fn a_gunnar_refusal_keeps_its_own_words() {
let fail = Fail::from(Status::not_found("repository \"a/b\" is not here"));
assert_eq!(fail.code, "notfound");
assert_eq!(fail.message, "repository \"a/b\" is not here");
assert_eq!(fail.http_status(), http::StatusCode::NOT_FOUND);
}
#[test]
fn an_unimplemented_rpc_is_not_a_server_error() {
// All of Mirror answers this on a real gunnar, and the console renders
// it as a disabled panel, so the status has to be distinguishable from
// a fault.
//
// Policy used to be on that list and is NOT any more: gunnar serves all
// six Policy RPCs off a document at `refs/gunnar/policy`, and the
// console now calls them (`api::policy`). The comment outlived the
// fact, which is how the branch-protection panel stayed disabled long
// after the server had grown the feature. `Refs.Diff` is likewise
// answered by the console's own diff engine rather than by gunnar.
let fail = Fail::from(Status::unimplemented("mirroring is not built"));
assert_eq!(fail.http_status(), http::StatusCode::NOT_IMPLEMENTED);
}
#[test]
fn a_refused_credential_is_reported_as_a_gateway_fault() {
// The operator's session is fine; the console's own identity is not.
// A 401 here would send the browser to a login page that cannot help.
let fail = Fail::from(Status::unauthenticated("no such key"));
assert_eq!(fail.http_status(), http::StatusCode::BAD_GATEWAY);
assert!(fail.is_console_identity_problem());
}
#[test]
fn a_clock_skew_refusal_is_temporary_and_not_an_identity_problem() {
// Measured wording, 2026-09-14. The key was fine; the appliance's clock
// was two hours ahead of the front's.
let fail = Fail::from(Status::unauthenticated(
"the credential is 7200691 ms away from this server's clock; the window is ±300000 ms",
));
assert!(fail.is_clock_skew());
assert!(!fail.is_console_identity_problem(), "a good key must not be reported as unregistered");
assert!(fail.is_infrastructure_fault());
assert_eq!(fail.http_status(), http::StatusCode::SERVICE_UNAVAILABLE);
}
#[test]
fn the_consoles_own_key_refused_is_maintenance_but_a_forge_token_rejection_is_not() {
// gunnar's own words for a console key it does not know (measured
// 2026-09-14, before the `monetize` principal was enrolled).
let ours = Fail::from(Status::unauthenticated(
"the credential did not verify against any registered key",
));
assert!(!ours.is_clock_skew());
assert!(ours.is_infrastructure_fault());
assert_eq!(ours.http_status(), http::StatusCode::SERVICE_UNAVAILABLE);
// A forge refusing the operator's token: theirs to fix, words kept.
let theirs = Fail::from(Status::unauthenticated("the token was rejected by github"));
assert!(!theirs.is_infrastructure_fault());
assert_eq!(theirs.http_status(), http::StatusCode::BAD_GATEWAY);
}
/// monetize's own operator sentence for the refusal the owner met on
/// 2026-09-20 (`Refusal::CannotAbsorb`, its Display), cut to the clause
/// that matters. Every word of it names the machine.
const CAPACITY_REASON: &str = "the product cannot deliver this to rickard: gunnar cannot serve \
100.0 GiB of stored bytes (pack_bytes + lfs_bytes): /data has 39.8 GiB free of 39.9 GiB, \
so 37.8 GiB may be served — short 62.1 GiB";
/// ★ **A capacity refusal after Buy renders the capacity reason** — the
/// sentence monetize wrote for the customer, carried on the metadata key,
/// and not the operator's message and not a word about maintenance.
#[test]
fn a_refusal_with_customer_words_shows_the_customer_words() {
let words = "This account cannot be given 100.0 GiB more right now: 37.8 GiB is the most \
that can be added. Nothing has been charged.";
let mut status = Status::failed_precondition(CAPACITY_REASON);
status.metadata_mut().insert_bin(
CUSTOMER_WORDS_KEY,
tonic::metadata::MetadataValue::from_bytes(words.as_bytes()),
);
let fail = Fail::from(status);
assert_eq!(fail.code, "failedprecondition");
assert_eq!(fail.message, words, "the customer's sentence must be the body");
assert!(
!fail.message.to_ascii_lowercase().contains("maintenance"),
"a full store is not maintenance: {}",
fail.message
);
assert!(fail.message.contains("Nothing has been charged"), "{}", fail.message);
assert_eq!(fail.http_status(), http::StatusCode::BAD_REQUEST);
}
/// ★ **An unknown refusal renders the server's own words.** No customer
/// sentence on the wire means the body is the message as monetize said it —
/// the console composes nothing in the gap, and above all not "maintenance".
#[test]
fn a_refusal_without_customer_words_is_shown_as_the_server_said_it() {
let fail = Fail::from(Status::failed_precondition(CAPACITY_REASON));
assert_eq!(fail.message, CAPACITY_REASON, "the server's words were replaced");
assert!(!fail.message.to_ascii_lowercase().contains("maintenance"), "{}", fail.message);
assert!(!fail.is_infrastructure_fault(), "a capacity refusal is not an estate fault");
assert_eq!(fail.http_status(), http::StatusCode::BAD_REQUEST);
}
/// An empty or unreadable customer sentence is no sentence: the server's
/// words stay. A key that is present but blank must not blank the body.
#[test]
fn blank_customer_words_do_not_blank_the_body() {
let mut status = Status::failed_precondition(CAPACITY_REASON);
status
.metadata_mut()
.insert_bin(CUSTOMER_WORDS_KEY, tonic::metadata::MetadataValue::from_bytes(b" "));
assert_eq!(Fail::from(status).message, CAPACITY_REASON);
let mut status = Status::failed_precondition(CAPACITY_REASON);
status
.metadata_mut()
.insert_bin(CUSTOMER_WORDS_KEY, tonic::metadata::MetadataValue::from_bytes(&[0xff, 0xfe]));
assert_eq!(Fail::from(status).message, CAPACITY_REASON);
}
#[test]
fn resource_exhausted_splits_size_from_rate() {
// gunnar sends one gRPC status for two unrelated limits; the console must
// not label a rate limit "payload too large". Both messages are gunnar's
// own (control_plane.rs for size, the admission gate for rate). Before the
// split BOTH answered 413 — this asserts they now diverge.
let too_big = Fail::from(Status::resource_exhausted(
"object 3f2a is 9000000 bytes, over this server's 4194304-byte ceiling \
for reading an object.",
));
assert_eq!(
too_big.http_status(),
http::StatusCode::PAYLOAD_TOO_LARGE,
"an oversized object is a 413"
);
let response_too_big = Fail::from(Status::resource_exhausted(
"object 3f2a is 9000000 bytes and the response would be 9000000, over the \
4194304-byte limit a gRPC client can receive by default.",
));
assert_eq!(
response_too_big.http_status(),
http::StatusCode::PAYLOAD_TOO_LARGE
);
let gate_full = Fail::from(Status::resource_exhausted(
"the admission gate is full: 8 transfers in flight at the server's limit; \
try again shortly",
));
assert_eq!(
gate_full.http_status(),
http::StatusCode::TOO_MANY_REQUESTS,
"a full admission gate is a 429, not a 413"
);
let write_budget = Fail::from(Status::resource_exhausted(
"this server holds a memory budget of 512 MiB and this push does not fit",
));
assert_eq!(
write_budget.http_status(),
http::StatusCode::TOO_MANY_REQUESTS
);
}
#[test]
fn a_denied_repository_is_not_blamed_on_the_consoles_key() {
// Observed against a real gunnar: asking for a repository that does not
// exist answers PermissionDenied, not NotFound. Treating that as a
// credential fault logs an error about a key that is working and shows
// the operator a gateway failure for a typo.
let fail = Fail::from(Status::permission_denied(
"principal gunnar-ui has no read grant on repo:vetra/nope",
));
assert_eq!(fail.http_status(), http::StatusCode::FORBIDDEN);
assert!(!fail.is_console_identity_problem());
}
#[tokio::test]
async fn a_cached_value_is_produced_once_within_its_window() {
let cache = Cached::new(Duration::from_secs(60));
let calls = std::sync::atomic::AtomicUsize::new(0);
let count = || async {
calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok::<_, ()>(7)
};
assert_eq!(cache.get(count()).await, Ok(7));
assert_eq!(cache.get(count()).await, Ok(7));
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[tokio::test]
async fn invalidating_forces_the_next_read_through() {
let cache = Cached::new(Duration::from_secs(60));
cache.get(async { Ok::<_, ()>(1) }).await.expect("first");
cache.invalidate().await;
assert_eq!(cache.get(async { Ok::<_, ()>(2) }).await, Ok(2));
}
#[tokio::test]
async fn a_failure_is_not_held() {
let cache: Cached<u8> = Cached::new(Duration::from_secs(60));
assert!(cache.get(async { Err::<u8, ()>(()) }).await.is_err());
assert_eq!(cache.get(async { Ok::<_, ()>(3) }).await, Ok(3));
}
#[tokio::test]
async fn an_address_without_a_scheme_still_connects() {
let signer = Arc::new(
Signer::new(ssh_key::PrivateKey::from(
ssh_key::private::Ed25519Keypair::from_seed(&[3u8; 32]),
))
.expect("a fixed seed makes a signer"),
);
assert!(Gunnar::connect_lazy("127.0.0.1:50051", signer).is_ok());
}
// ── a main that reboots under the console (row recharge-main-front-cut-off) ──
/// How long the console may take to notice that its connection to a main
/// that rebooted is dead — before the next call is made on it.
const LIVENESS_WINDOW: Duration = Duration::from_secs(15);
#[test]
fn the_keepalive_fits_the_liveness_window() {
assert!(KEEPALIVE_EVERY + Duration::from_secs(1) < LIVENESS_WINDOW, "a dead connection is pinged, and so noticed, inside the window");
}
/// **A busy main is not a dead one** (R5.1-G, 2026-10-07). gunnar's
/// control plane answered a PING from the same workers that ran its slow
/// synchronous calls (a snapshot, a disk walk), so a burst of them left a
/// PING unanswered for seconds; with five seconds the console killed the
/// connection and every call in flight on it — test2's `KeepAliveTimedOut`
/// stalls, a page that failed while main was merely busy. gunnar now
/// answers PINGs from a runtime no call can hold, and the console gives a
/// PING 20–30 s anyway, for a main that is not upgraded yet and for a
/// loaded box. A main that reboots is still noticed at once (its RST, the
/// test below); one gone in silence after `KEEPALIVE_EVERY + KEEPALIVE_TIMEOUT`.
#[test]
fn a_busy_main_is_given_time_to_answer_a_ping() {
assert!(
KEEPALIVE_TIMEOUT >= Duration::from_secs(20) && KEEPALIVE_TIMEOUT <= Duration::from_secs(30),
"the console gives a PING {KEEPALIVE_TIMEOUT:?}: under 20 s a busy main's connection is killed with \
every call on it, over 30 s a main that is gone in silence is noticed too late"
);
}
fn console_signer() -> Arc<Signer> {
Arc::new(
Signer::new(ssh_key::PrivateKey::from(
ssh_key::private::Ed25519Keypair::from_seed(&[5u8; 32]),
))
.expect("a fixed seed makes a signer"),
)
}
/// ★ **GHUNT-19: a person's forge event leaves the console UNSIGNED**, from
/// their own session, for gunnar to attest after a fresh touch; the console's
/// own client still signs with its key. Before, the console signed a
/// person's approval with ITS key in their name, and gunnar refused it.
#[tokio::test]
async fn a_persons_forge_event_is_unsigned_and_the_consoles_is_signed() {
let gunnar = Gunnar::connect_lazy("http://127.0.0.1:9", console_signer()).expect("a lazy channel");
let event = crate::forge::ForgeEvent {
repo: "acme/app".into(),
pull: 1,
actor: "anna".into(),
at_unix_ms: 1_000,
event: crate::forge::Event::Approved { commit: "c0ffee".into() },
};
let hers = gunnar.as_session("annas-session").sign_forge_event(&event).expect("an event");
assert!(
hers.key_fingerprint.is_empty() && hers.signature.is_empty(),
"a person's event must not carry the console's signature in her name: {}",
hers.key_fingerprint
);
assert_eq!(hers.event.as_ref().map(|e| e.actor.as_str()), Some("anna"));
assert!(hers.attested.is_none(), "an attestation is gunnar's to make, never the console's");
let console = gunnar.sign_forge_event(&event).expect("signed");
assert_eq!(console.key_fingerprint, gunnar.fingerprint());
assert!(!console.signature.is_empty());
}
/// gunnar on a box where nothing happens: `Events.Watch` says its prologue
/// and then nothing, for as long as the stream is held; `Bootstrap.Status`
/// answers.
#[derive(Clone, Default)]
struct QuietMain;
#[tonic::async_trait]
impl crate::pb::events_server::Events for QuietMain {
async fn watch(&self, _: Request<crate::pb::WatchRequest>) -> Result<tonic::Response<tonic::codegen::BoxStream<crate::pb::Event>>, Status> {
use tokio_stream::StreamExt as _;
let opened = crate::pb::Event {
payload: Some(crate::pb::event::Payload::StreamOpened(crate::pb::StreamOpened { epoch: "run-1".into(), ..Default::default() })),
..Default::default()
};
Ok(tonic::Response::new(Box::pin(tokio_stream::iter(vec![Ok(opened)]).chain(tokio_stream::pending()))))
}
}
#[tonic::async_trait]
impl crate::pb::bootstrap_server::Bootstrap for QuietMain {
async fn status(&self, _: Request<crate::pb::BootstrapStatusRequest>) -> Result<tonic::Response<crate::pb::BootstrapStatusResponse>, Status> {
Ok(tonic::Response::new(crate::pb::BootstrapStatusResponse::default()))
}
}
async fn serve_quiet_main() -> std::net::SocketAddr {
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = l.local_addr().unwrap();
tokio::spawn(
tonic::transport::Server::builder()
.add_service(crate::pb::events_server::EventsServer::new(QuietMain))
.add_service(crate::pb::bootstrap_server::BootstrapServer::new(QuietMain))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(l)),
);
at
}
/// **The utility network between the front and the main, and the main's
/// kernel.** Bytes are carried both ways until the main REBOOTS (a new boot
/// number on the sender): then every connection it had is forgotten in
/// silence — no FIN, no RST, as a hard reset forgets them — and the front's
/// next segment on one of them is answered with RST, as a kernel that never
/// heard of the connection answers it. A connection made after the reboot
/// reaches the main again.
async fn fabric(main: std::net::SocketAddr) -> (String, tokio::sync::watch::Sender<u32>) {
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = format!("http://{}", l.local_addr().unwrap());
let (boots, boot) = tokio::sync::watch::channel(0u32);
tokio::spawn(async move {
while let Ok((front, _)) = l.accept().await {
tokio::spawn(carry(front, main, boot.clone()));
}
});
(at, boots)
}
async fn carry(front: tokio::net::TcpStream, main: std::net::SocketAddr, mut boot: tokio::sync::watch::Receiver<u32>) {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
boot.borrow_and_update();
let Ok(to_main) = tokio::net::TcpStream::connect(main).await else { return };
let (mut fr, mut fw) = front.into_split();
let (mut mr, mut mw) = to_main.into_split();
let (mut a, mut b) = (vec![0u8; 1 << 16], vec![0u8; 1 << 16]);
loop {
tokio::select! {
n = fr.read(&mut a) => match n {
Ok(n) if n > 0 => if mw.write_all(&a[..n]).await.is_err() { return },
_ => return,
},
n = mr.read(&mut b) => match n {
Ok(n) if n > 0 => if fw.write_all(&b[..n]).await.is_err() { return },
_ => return,
},
_ = boot.changed() => break,
}
}
// The main rebooted: its side of this connection is gone.
drop((mr, mw));
// The front's next segment reaches a kernel that never heard of it.
let _ = fr.read(&mut a).await;
if let Ok(front) = fr.reunite(fw) {
let _ = front.set_zero_linger();
}
}
/// **A main that reboots in silence does not cost the console its next
/// call.** MEASURED on the KVM mock C, 2026-09-30 (row
/// `recharge-main-front-cut-off`): the recharged main, restored from its
/// twin, REBOOTED at 22:28:53 (a hard reset: XFS recovery on its disks at
/// the next boot). The console's channel had re-subscribed to its event
/// stream at 22:28:52; nothing flowed on it for thirty minutes, nothing was
/// written into the dead connection and nothing noticed it — no "the event
/// stream from gunnar ended" until 22:59:35, when the functest's
/// `POST /api/namespaces` was written into it, the main's new kernel
/// answered RST, and the console served `HTTP 502 transport error`.
#[tokio::test]
async fn a_main_that_reboots_in_silence_does_not_cost_the_next_call() {
let (fabric, boots) = fabric(serve_quiet_main().await).await;
let gunnar = Gunnar::connect_lazy(&fabric, console_signer()).expect("an address");
let mut events = gunnar.events().watch(crate::pb::WatchRequest::default()).await.expect("subscribed").into_inner();
assert!(events.message().await.expect("the prologue").is_some(), "the prologue arrives");
// Quiet, as the thirty minutes were: whatever the front had still to
// say about the prologue (flow-control credit) is said before the main
// goes.
tokio::time::sleep(Duration::from_secs(1)).await;
boots.send(1).expect("the main reboots");
// The console holds its subscription, as the EventHub does, for the
// window in which the dead connection must be noticed.
let noticed = tokio::time::timeout(LIVENESS_WINDOW, events.message()).await;
let call = gunnar.bootstrap().status(crate::pb::BootstrapStatusRequest::default()).await;
assert!(
call.is_ok(),
"the first call after the main rebooted must reach the main again, not the connection its old kernel forgot (the functest's HTTP 502): {:?}",
call.err()
);
assert!(matches!(noticed, Ok(Err(_))), "the idle event stream must notice its dead connection within {LIVENESS_WINDOW:?}: {noticed:?}");
}
}