nordisk/gunnar
crates/gunnar-server/src/serve.rs
//! `gunnar serve` starts its listeners together and stops them together. They
//! share nothing but the control plane behind them.
//!
//! ```text
//! :2222 SSH git-upload-pack / git-receive-pack, publickey auth
//! :8080 HTTP the same two verbs over smart HTTP, access-token
//! auth — only with `--features http` AND `--http-bind`
//! 127.0.0.1:50051 gRPC the control plane; authenticated (sshsig + the
//! ACL model) unless the development escape hatch is on
//! ```
//!
//! **The two git listeners are two doors onto one room**, and the wiring below
//! is what keeps that true: the same `Dispatch`, the same `WireTransport`, the
//! same `ServerMetrics` and the one `allow_anonymous` field are handed to both.
//! A second control plane or a second transport here is how SSH and HTTP would
//! come to disagree about who may push.
//!
//! Git traffic never passes through gRPC. There is no bridge and none is
//! wanted: it is that separation which makes the v1 authentication hole
//! survivable, because an open gRPC port cannot push code.
//!
//! # What this module is responsible for saying out loud
//!
//! 1. **Whether the unauthenticated enrolment door is open**, at every start,
//! on stdout *and* in the log. `gunnar_core::AuthTable::prepare_to_serve`
//! returns the state as a value; [`announce_readiness`] is the caller that
//! has to branch on it, and `tests/bootstrap.rs` asserts every branch.
//!
//! Until 2026-08-05 an empty identity table with no `--admin-key` was a
//! **refusal to start** (the secret-store seam). That is why gunnar had no first boot: the
//! box could not be reached to be given its first key, and the only way in
//! was `gunnar key add` with the server down, which needs a shell the
//! appliance does not have. It now starts and serves `Bootstrap.Enroll`
//! until somebody uses it — see `gunnar_grpc::service::bootstrap`.
//! 2. **gRPC binds loopback unless told otherwise**, and says so at startup
//! whatever it binds.
//!
//! # Two listeners, and now two runtimes
//!
//! Separate listeners separate the *ports*. They did not separate the *schedulers*, and that
//! turned out to be most of what the separation was for.
//!
//! `ControlPlane` is a synchronous trait — `gunnar-grpc`'s `seam.rs` says so,
//! and says the backend belongs behind `spawn_blocking` "at the call site" —
//! and no call site does it. `Admin.StorageInfo` runs an unbounded recursive
//! `std::fs::read_dir` that `stat`s every loose object in the repository,
//! straight from its `async fn`, on a worker thread. `Repos.Create` opens a git
//! directory and replays an arrow log. While both listeners shared one runtime,
//! a handful of concurrent control-plane calls occupied every worker and the SSH
//! listener's accept, handshake and exec dispatch simply did not run. The loopback default leaves
//! the control port unauthenticated, so that was reachable on purpose as well as
//! by accident.
//!
//! **The control plane therefore gets its own runtime**, built here, with its
//! own worker threads and its own blocking pool. Two properties follow that
//! thirty `spawn_blocking` wrappers would not have given:
//!
//! * it holds for *every* control-plane call, including ones written later by
//! somebody who never read `seam.rs`. The fix is by construction rather than a
//! convention thirty call sites have to keep (one implementation);
//! * a control plane that stalls completely degrades the control plane only. Git
//! keeps being served.
//!
//! The gRPC listener is bound **on** that runtime rather than handed to it: a
//! tokio socket registers with the I/O driver of the runtime that created it,
//! and one whose driver belongs to a runtime that is shutting down is a stall
//! waiting to happen. `tests/control_plane_starvation.rs` measures the whole
//! thing from an OS thread outside both runtimes.
use std::net::SocketAddr;
use std::sync::Arc;
use anyhow::{bail, Context, Result};
use gunnar_core::observe::{MetricsSnapshot, RequestId, ServerMetrics};
use gunnar_core::{ensure_host_key, Readiness};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_ssh::{BoundServer, GunnarSsh, SshConfig};
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use gunnar_common::bindspec::BindSpec;
use crate::control_plane::CoreControlPlane;
use crate::stores::ServerStores;
use crate::transport::{WireConfig, WireTransport};
/// What `gunnar serve` was asked for.
#[derive(Debug, Clone)]
pub struct ServeConfig {
/// Where the SSH listener binds.
pub ssh_bind: SocketAddr,
/// Where the gRPC control plane binds. Loopback by default.
///
/// A [`BindSpec`] rather than a `SocketAddr` because an appliance's private
/// address is DHCP-learned: the declaration names `eth1:50051` and
/// [`run`] resolves it, ONCE, after `GUNNAR_NET` has brought the link up
/// and before anything binds. Downstream — the readiness line, the
/// dashboard row, `Admin.Health` — then reports the address the listener
/// actually took, not the spelling it was asked for.
pub grpc_bind: BindSpec,
/// The DEVICE the control plane listens on, when `grpc_bind` was spelled as
/// an interface (`eth1:50051`) — set by [`run`]/[`start`] from that spelling
/// and honoured as `SO_BINDTODEVICE` on the listener socket, so the
/// interface name means the interface and not merely the address it held
/// at boot. MEASURED 2026-09-05 (the KVM bench rig, two slirp NICs on the
/// same 10.0.2.0/24): an address-only bind of eth1's 10.0.2.15 answered
/// through eth0, because both NICs held that address and the route decided.
/// On a cloud whose utility NIC is `eth1` on its own subnet the two agree;
/// the device bind makes them agree everywhere. `None` for an address spec.
pub grpc_bind_device: Option<String>,
/// An OpenSSH public key line to install as the bootstrap admin when the
/// identity table is empty (the secret-store seam).
pub admin_key: Option<String>,
/// **The keys `Bootstrap.Restore` is taken from** (`--restore-pubkey`, h-v1-6a): OpenSSH
/// public key lines or `@<path>`, resolved and checked at start. An installer-fed box adds
/// the fed admin keys itself; a box with neither refuses every restore by name.
pub restore_pubkeys: Vec<String>,
/// Start the SSH listener only.
pub no_grpc: bool,
/// **`--wg-only`: every box-to-box connection runs over the estate's
/// WireGuard overlay** (owner, 2026-10-04). The address the control plane
/// is bound on (`eth1:50051`, the utility network) serves NOTHING: the
/// control plane and the replication listener answer only on the tunnel's
/// address ([`crate::tunnel_addr`]), and SSH — bound for the public's git
/// clients — drops a connection that arrives at the utility address before
/// a byte is spoken. A plain-IP connection between two boxes is refused.
pub wg_only: bool,
/// `--role twin`: no forge listener from the first boot (loopback only),
/// whether or not `Bootstrap.BecomeTwin` has run yet.
pub twin_role: bool,
/// Whether a caller with **no key** may clone a repository marked
/// [`Visibility::Public`](gunnar_core::Visibility::Public). `true` by
/// default; nothing is anonymously readable until a repository is
/// deliberately made public, because `Private` is the default for a new
/// one.
///
/// Reaches [`SshConfig::allow_anonymous`](gunnar_ssh::SshConfig::allow_anonymous)
/// **and `HttpConfig::allow_anonymous`**, from this one field. Two fields
/// is how one of them ends up stubbed: an operator who passed
/// `--no-anonymous` and got it honoured on one transport has a server that
/// serves strangers and reports that it does not. The gRPC control plane
/// has no anonymous mode to switch, at any setting.
pub allow_anonymous: bool,
/// **`--anonymous allow|refuse`** (the browse replica): whether an
/// anonymous read is SERVED here or sent to the browse replica the
/// refusal names. Orthogonal to `allow_anonymous`, which is whether a
/// stranger authenticates at all; this is what the dispatcher tells one
/// who did. Both front doors read it through the one `Control`.
pub anonymous: gunnar_ssh::AnonymousPolicy,
/// Where the smart-HTTP listener binds. `None` does not start
/// it, which is the default.
///
/// Off unless asked for, because a second listening port is a decision an
/// operator makes and not one a default makes for them. The feature it
/// needs is off by default too, so an appliance image compiles no HTTP at
/// all.
#[cfg(feature = "http")]
pub http_bind: Option<SocketAddr>,
/// Whether the HTTP listener will read a credential off a plain connection
/// Default: only when a proxy attests TLS.
#[cfg(feature = "http")]
pub http_cleartext: gunnar_http::Cleartext,
/// Terminate TLS on the smart-HTTP listener itself. `None` — the default —
/// is the proxy-terminated shape; `SelfMinted` is the single-executable
/// shape (leaf minted from this server's own issuer at start, chain
/// closed for clients that pin the issuer); `Identity` is
/// bring-your-own-PEM.
#[cfg(feature = "http")]
pub http_tls: Option<HttpTls>,
// (HttpTls is defined right below ServeConfig.)
/// Require a **client certificate** on the gRPC control plane, minted by
/// this server's own issuer (the mTLS plan).
///
/// `None` — the default — is the loopback shape the unauthenticated v1 assumes: the listener
/// binds `127.0.0.1` and a credential in metadata is the identity.
/// `Some(GrpcTls::SelfMinted)` is the shape a control plane reachable off
/// the box needs, and it changes two things at once: the transport refuses
/// a peer whose certificate this issuer did not sign, and the gate maps the
/// certificate that got through to an ACL principal.
///
/// Deliberately **not** on by default. Turning it on rewrites how every
/// existing client must connect, and a security default that silently
/// breaks the working configuration gets switched back off in a hurry —
/// which leaves it off everywhere. It is a flag an operator turns on when
/// the listener leaves loopback, and `--grpc-bind` off loopback without it
/// is what the startup warning names.
pub grpc_tls: Option<GrpcTls>,
/// `--web-relying-party`, unparsed: the WebAuthn relying parties the gRPC
/// plane answers for (`gunnar_grpc::webauthn::RelyingParty`).
pub web_relying_parties: Vec<String>,
/// `--web-require-user-verification` (A34-12).
pub web_require_user_verification: bool,
/// `--identity-root-site` (IDENTITY-HOST): sign-up is open here.
pub identity_root_site: bool,
/// **The mutual replication listener** (`--grpc-replication-bind`, lane T9):
/// the door a twin follows this box through. Served only with a client
/// certificate this box's issuer signed, TLS 1.3 only, its leaf minted from
/// the issuer at every start under the fixed name
/// `gunnar_grpc::client_signer::REPLICATION_SERVER_NAME`.
///
/// `None` DERIVES it, and the derivation is what lets the ONE medium carry
/// no flag: a control plane bound off loopback (the appliance,
/// `0.0.0.0:50051`) gets this listener on the same address, port + 1
/// (`50052`); a loopback one (a laptop, a test) gets none. Its main
/// listener stays exactly what it was, because turning `--grpc-tls-self` on
/// for the whole plane would lock out every client that has no certificate
/// — and the bootstrap door with them.
pub grpc_replication_bind: Option<SocketAddr>,
// (HttpTls is defined right below ServeConfig.)
/// How many git transfers may be in flight across the whole server.
/// `None` takes
/// [`DEFAULT_MAX_CONCURRENT_TRANSFERS`](gunnar_ssh::config::DEFAULT_MAX_CONCURRENT_TRANSFERS).
///
/// The fix note on the admission gate says every one of these numbers needs to be
/// settable, because the right one is deployment-specific; until now
/// neither was reachable from `gunnar serve` and the two defaults were the
/// only values any deployment could have. That is also why the pressure
/// harness could not reach contention without 320 client processes.
pub max_concurrent_transfers: Option<usize>,
/// The **memory budget** the admission ledger may give out to git
/// operations in flight, in bytes (gate 2). `None` takes
/// [`crate::budget::budgets`]'s derived figure — `MemAvailable` x
/// [`crate::budget::TRANSFER_MEMORY_PERCENT`] — rather than
/// `gunnar-ssh`'s compiled-in 8 GiB, which was a number about the cgroup
/// one bakeoff ran in and not about this box.
pub memory_budget_bytes: Option<u64>,
/// How long a transfer may **wait** for a slot before it is refused.
/// `None` takes [`gunnar_ssh::governor::DEFAULT_ADMISSION_WAIT`];
/// `Some(Duration::ZERO)` restores the instant refusal.
pub admission_wait: Option<std::time::Duration>,
/// `--transfer-idle-secs` (GUN-SSH-SLOT-HANG): `None` keeps the SSH default (180 s);
/// zero turns it off.
pub transfer_idle: Option<std::time::Duration>,
/// A principal's guaranteed floor and its share of a contended server.
/// `None` takes
/// [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`](gunnar_ssh::config::DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL).
pub max_transfers_per_principal: Option<usize>,
/// How many git reads may be served **without a key** at once.
/// `None` takes
/// [`DEFAULT_MAX_ANONYMOUS_TRANSFERS`](gunnar_ssh::admission::DEFAULT_MAX_ANONYMOUS_TRANSFERS).
///
/// Note the asymmetry with the field above, which is deliberate: that one
/// is a floor a principal is guaranteed and may exceed on a quiet box, this
/// one is a wall that binds however empty the server is. See
/// [`gunnar_ssh::admission`] for why work-conserving is the right instinct
/// for a tenant and the wrong one for a stranger.
pub max_anonymous_transfers: Option<usize>,
/// **The usage share** (`admit-share`): `--admit-limit` and
/// `--admit-window-days`, parsed. See [`gunnar_ssh::SshConfig::admit`].
#[cfg(feature = "admit-share")]
pub admit: gunnar_ssh::edda_admit::Settings,
/// Make the per-principal number an unconditional wall again rather than a
/// max-min share. See
/// [`SshConfig::per_principal_limit_is_hard`](gunnar_ssh::SshConfig::per_principal_limit_is_hard).
pub per_principal_limit_is_hard: bool,
/// How many worker threads the control plane's **own** runtime gets.
/// `None` derives it from [`control_plane_workers`].
///
/// A knob rather than a constant, because the right number is a deployment
/// question: a box whose operators live in the UI wants more, an appliance
/// serving one CI fleet wants fewer. A test also has to be able to make the
/// pool small enough to saturate it deliberately.
pub control_plane_workers: Option<usize>,
/// The largest object `Refs.ReadObject` will inflate. `None`
/// takes
/// [`CoreControlPlane`](crate::CoreControlPlane)'s default of 64
/// MiB; `Some(0)` means no ceiling.
///
/// A knob for the same reason `control_plane_workers` is one: the number is
/// `memory budget / concurrent inflates`, and the budget is a deployment
/// question. It is also the only one of these that turns a request which
/// works today into a refusal, which is exactly why it must be reachable
/// rather than compiled in.
pub max_object_bytes: Option<u64>,
/// How often the counter heartbeat is written. `None` turns it
/// off.
///
/// The heartbeat answers *"what is this server doing right now"* from a log
/// the operator already has open, with nothing to scrape. It stays quiet
/// while nothing changes, so an idle appliance does not fill its journal.
pub metrics_interval: Option<std::time::Duration>,
/// Where the **scrape endpoint** binds. `None` does not start it,
/// which is the default.
///
/// The heartbeat above and this are the same nineteen numbers reached two
/// ways, and both are wanted: the heartbeat needs nothing set up and is
/// already in the journal when an incident is reconstructed afterwards; the
/// endpoint can be *asked*, at whatever rate the person asking chooses,
/// which is the only way `transfers_in_flight` distinguishes a slow
/// transfer from a stuck one while it is still happening.
///
/// Off unless asked for, for the same reason [`ServeConfig::http_bind`] is:
/// a second listening port is an operator's decision. See
/// [`crate::scrape`] for why it carries no authentication and what
/// that does and does not expose.
pub metrics_bind: Option<SocketAddr>,
/// Replication to a second gunnar (replication). `None` reads
/// `<data-root>/replication.json`, which is what an operator configures and
/// what a restart picks up; `Some` overrides it in-process and is how a
/// test points a primary at a standby it just started on an ephemeral port.
///
/// A configuration with no targets starts nothing at all: no task, no
/// subscriber. A server that replicates nowhere pays nothing for the
/// feature existing.
#[cfg(feature = "replication")]
pub replication: Option<crate::mirror::ReplicationConfig>,
/// **I11's passive heartbeat**: probe this primary and take over if it stops
/// answering. `None` — the default — starts no task at all.
///
/// Off unless asked for, and that is the design rather than caution:
/// `control_plane.rs::replication_abdicate` states the rule, *"naming a new
/// primary is the operator's decision, never a default this box picks"*, and
/// a box that promoted itself because nobody configured it not to would
/// make that decision for every deployment that never heard of the feature.
///
/// It does nothing on a box whose role is `Primary`: the task re-reads the
/// role every tick and leaves when it is not a standby, so setting it on
/// both boxes of a pair is harmless and is what a symmetric unit file wants.
/// See [`crate::standby_watch`] for what the probe does and does not prove.
#[cfg(feature = "replication")]
pub standby_watch: Option<crate::standby_watch::StandbyWatch>,
/// **The estate repairs itself** (5.4): the boot check after the control plane
/// listens (it RECHARGES only with `auto`, the estate's opt-in, off by default)
/// and the pull channel to monetize. `None` (the default): neither runs.
/// See [`crate::repair`].
#[cfg(feature = "replication")]
pub repair: Option<crate::repair::RepairConfig>,
/// **Where replication's counters are published to**, for the console band, the
/// staleness alarm and `Admin.Stats`.
///
/// It is on the config rather than returned from [`Server::start`] because
/// the console band starts BEFORE the server does — [`run`] hands the same
/// view to both, and `start` drops the replicator into it once there is
/// one. Left at its default it reports "nothing configured", which is what
/// a server with no standby is.
pub replication_view: crate::replication_view::ReplicationView,
/// **Where the band's `tenants` row is read from**, the same shape and
/// for the same reason as `replication_view`: a metered `Server::start`
/// fills it from its entitlement slot; a community build never does, and
/// the band draws no row.
pub tenants_view: crate::dashboard::TenantView,
/// **Where the band's `flow` row is read from.** Filled on every build —
/// counting what moved is not the metered appliance's capability. See
/// [`crate::dashboard::FlowView`].
pub flow_view: crate::dashboard::FlowView,
/// What the two git state machines serve under (`gunnar-wire`'s own
/// `Config` types, held once — see [`WireConfig`]).
///
/// Defaults to `gunnar-wire`'s defaults, which is what every deployment
/// silently ran on while `with_config` had no callers. The knobs it carries
/// were fully implemented in both protocol versions and reachable from
/// nowhere; `--no-filter` in particular is the difference between an
/// appliance that declines partial clone and one that only says it might.
pub wire: WireConfig,
/// **The metered appliance's key and pay page** (`metered` only): what
/// `--monetize-pubkey` and `--billing-url` resolve to. `None` on a metered
/// build starts it with every namespace Free and `Entitlement.Set` refusing
/// by name — never with a key it did not have.
#[cfg(feature = "metered")]
pub metered: Option<crate::entitlement::MeteredConfig>,
/// **The disk-headroom floor** (`--disk-floor`): the community plan gate,
/// first link of every front door's chain and the guard on `Repos.Create`.
/// See `gunnar_ssh::headroom`. Always present — the default is 5 % or
/// 2 GiB, whichever is larger — because a box with no floor is a box that
/// finds out it is full from a torn pack.
pub disk_floor: gunnar_ssh::DiskFloor,
/// **What measures the data set** for the floor and the space ledger.
/// `None` — every deployment — is one `statvfs` on the data root. A test
/// that must fill a data set of a stated size without a filesystem of that
/// size hands its own probe here (`tests/push_space_admission.rs`'s
/// virtual disk); nothing else about the server changes.
pub space_probe: Option<Arc<dyn gunnar_ssh::SpaceProbe>>,
}
/// The `--anonymous` policy type, re-exported so a caller that builds a
/// [`ServeConfig`] (the test harness, `gunnar-cli`) names it from here.
pub use gunnar_ssh::AnonymousPolicy;
impl Default for ServeConfig {
fn default() -> Self {
ServeConfig {
ssh_bind: SocketAddr::from(([0, 0, 0, 0], gunnar_ssh::config::DEFAULT_PORT)),
grpc_bind: BindSpec::Addr(gunnar_grpc::DEFAULT_BIND),
grpc_bind_device: None,
admin_key: None,
restore_pubkeys: Vec::new(),
no_grpc: false,
wg_only: false,
twin_role: false,
allow_anonymous: true,
anonymous: gunnar_ssh::AnonymousPolicy::Allow,
#[cfg(feature = "http")]
http_bind: None,
#[cfg(feature = "http")]
http_cleartext: gunnar_http::Cleartext::default(),
#[cfg(feature = "http")]
http_tls: None,
grpc_tls: None,
web_relying_parties: Vec::new(),
identity_root_site: false,
web_require_user_verification: false,
grpc_replication_bind: None,
max_concurrent_transfers: None,
memory_budget_bytes: None,
admission_wait: None,
transfer_idle: None,
max_transfers_per_principal: None,
max_anonymous_transfers: None,
#[cfg(feature = "admit-share")]
admit: gunnar_ssh::edda_admit::Settings::default(),
per_principal_limit_is_hard: false,
control_plane_workers: None,
max_object_bytes: None,
metrics_interval: Some(METRICS_INTERVAL),
metrics_bind: None,
#[cfg(feature = "replication")]
replication: None,
#[cfg(feature = "replication")]
standby_watch: None,
#[cfg(feature = "replication")]
repair: None,
replication_view: crate::replication_view::ReplicationView::new(),
tenants_view: crate::dashboard::TenantView::new(),
flow_view: crate::dashboard::FlowView::new(),
wire: WireConfig::default(),
#[cfg(feature = "metered")]
metered: None,
disk_floor: gunnar_ssh::DiskFloor::default(),
space_probe: None,
}
}
}
impl ServeConfig {
/// A configuration whose listeners take kernel-assigned loopback ports.
/// A constructor for tests, not a mode: nothing about what is authenticated
/// or dispatched changes.
pub fn ephemeral() -> Self {
ServeConfig {
ssh_bind: SocketAddr::from(([127, 0, 0, 1], 0)),
grpc_bind: BindSpec::Addr(SocketAddr::from(([127, 0, 0, 1], 0))),
grpc_bind_device: None,
..ServeConfig::default()
}
}
}
/// How long [`Server::shutdown`] lets the gRPC listener finish gracefully
/// before it is aborted.
///
/// Long enough for an in-flight unary call to complete on a loaded box, short
/// enough that a Ctrl-C is answered while the operator is still watching. See
/// [`Server::shutdown`] for why any bound is needed at all.
pub const GRPC_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
/// How often `gunnar serve` writes its counters to the log.
///
/// Sixty seconds: often enough that an operator watching an incident sees the
/// numbers move while they are still watching, rare enough that a year of
/// journal on an idle appliance is not made of heartbeats. The line is skipped
/// when nothing has changed since the last one, so "quiet" and "not running"
/// stay distinguishable. A *busy* server is never quiet.
pub const METRICS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
/// The size of the blocking pool that carries git transfers.
///
/// **A decision, not a default.** Every in-flight transfer occupies one blocking
/// thread for its whole duration (`WireTransport::serve` bridges the async
/// duplex into `gunnar-wire`'s blocking `BufRead`/`Write` through
/// `SyncIoBridge`), so this number *was* the server's real concurrency ceiling,
/// and it was 512 because that is what `tokio` picks when nobody says. Past it,
/// transfers queued with no log, no counter and no error.
///
/// The ceiling that matters is now
/// [`SshConfig::max_concurrent_transfers`](gunnar_ssh::SshConfig::max_concurrent_transfers),
/// which refuses visibly. This pool has to stay comfortably **above** it: it
/// also carries every other blocking thing the server does — pack indexing, ref
/// transactions, the arrow log — and a pool sized exactly to the transfer limit
/// would deadlock the moment a transfer needed a second blocking task.
/// `the_blocking_pool_is_larger_than_the_transfer_limit` asserts the ordering
/// rather than leaving it to this paragraph.
pub const GIT_BLOCKING_THREADS: usize = 512;
/// Worker threads that RUN the control plane's calls, derived from the
/// machine.
///
/// A quarter of the cores, at least four and at most eight. The control plane is
/// an administrative surface — a handful of operators and a UI — not the
/// serving path, and the whole point of giving it a separate pool is that stalls
/// stay inside a bounded number of threads. Giving it one core's worth would
/// make a single slow `StorageInfo` serialise the UI; giving it the machine
/// would recreate the problem with extra steps. The floor was two until
/// R5.1-G (2026-10-07): two slow synchronous calls held both, and every other
/// call — the console's pages, monetize's `Entitlement.Set` — queued behind
/// them.
///
/// These are the HANDLER workers ([`ServiceContext::with_handler_runtime`]).
/// The HTTP/2 connections run on [`CONTROL_CONNECTION_WORKERS`] workers of
/// their own, which no handler can hold, so a client's keepalive PING is
/// answered however long the calls take.
pub fn control_plane_workers() -> usize {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
(cores / 4).clamp(4, 8)
}
/// Worker threads that drive the control plane's HTTP/2 connections (R5.1-G).
///
/// They accept, read and write frames — and answer a client's keepalive PING —
/// and nothing else: every call runs on the handler runtime
/// ([`control_plane_workers`] workers). They used to be one runtime, and when
/// the slow synchronous calls held all of its workers nothing answered the
/// PING; the console and monetize then killed the connection (their keepalive:
/// a PING every 10 s, dead 5 s later) and every call in flight on it failed.
pub const CONTROL_CONNECTION_WORKERS: usize = 2;
/// How many CPUs this process may **run on**, from its affinity mask.
///
/// `/proc/self/status`'s `Cpus_allowed_list`, because that is the kernel's own
/// answer and not a request anybody made. `Some(n)` or `None` when the file is
/// not there to be read; a missing answer is never guessed at, because the whole
/// point of [`report_cpu_budget`] is a comparison and half a comparison is
/// worse than none.
fn cpus_allowed() -> Option<usize> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
let list = status
.lines()
.find_map(|line| line.strip_prefix("Cpus_allowed_list:"))?;
count_cpu_list(list)
}
/// How many CPUs a kernel CPU list names: `0-31`, `3`, `0-3,8,10-11`.
///
/// Split out of [`cpus_allowed`] so it can be exercised on lists this host does
/// not have. It matters: the reference desktop reports a plain `0-31`, so a version of this that
/// counted ranges and dropped the singletons passed every test on this box
/// while answering **0 allowed CPUs on a server pinned to one core** — the
/// condition the caller exists to detect, read exactly backwards.
fn count_cpu_list(list: &str) -> Option<usize> {
let mut total = 0usize;
for range in list.trim().split(',') {
let range = range.trim();
if range.is_empty() {
continue;
}
total += match range.split_once('-') {
Some((lo, hi)) => {
let (lo, hi) = (
lo.trim().parse::<usize>().ok()?,
hi.trim().parse::<usize>().ok()?,
);
hi.checked_sub(lo)?.checked_add(1)?
}
None => {
range.parse::<usize>().ok()?;
1
}
};
}
(total > 0).then_some(total)
}
/// Whether the affinity is wide enough against the quota to be worth a line,
/// and the two numbers to say it with.
///
/// Pure, and given both readings rather than taking them, so the condition can
/// be seen in both states from a test on any machine. Either reading missing is
/// `None`: half a comparison is worse than none.
///
/// **Twice, not any excess.** The two numbers are rarely exactly equal and a
/// line that fires on nine-versus-eight is a line operators learn to skip.
/// Twice is where the measurement above already showed a real cost.
fn spread_budget(allowed: Option<usize>, quota: Option<f64>) -> Option<(usize, usize)> {
let (allowed, quota) = (allowed?, quota?);
let budget = quota.ceil().max(1.0) as usize;
(allowed >= budget.saturating_mul(2)).then_some((allowed, budget))
}
/// How many CPUs' worth of time this process may **use**, from cgroup v2
/// `cpu.max`. `None` when there is no quota, which is the ordinary case.
fn cpu_quota() -> Option<f64> {
let own = std::fs::read_to_string("/proc/self/cgroup").ok()?;
let rel = own
.lines()
.find_map(|line| line.strip_prefix("0::"))?
.trim()
.trim_start_matches('/');
let path = std::path::Path::new("/sys/fs/cgroup")
.join(rel)
.join("cpu.max");
let raw = std::fs::read_to_string(path).ok()?;
let mut parts = raw.split_whitespace();
let quota = parts.next()?;
if quota == "max" {
return None;
}
let quota: f64 = quota.parse().ok()?;
let period: f64 = parts.next()?.parse().ok()?;
(period > 0.0).then(|| quota / period)
}
/// **Say when the box will let this server run far wider than it will let it
/// work**, because that costs more than anything in the request path.
///
/// A container started with `--cpus 8` on a 32-core host gets eight CPUs of
/// *quota* spread over thirty-two CPUs of *affinity*. gunnar serves one transfer
/// on three threads that hand work to each other — a hyper task on a runtime
/// worker, the `spawn_blocking` thread that runs the git state machine, and the
/// pack-entry worker `iter_from_counts` spawns — so at N concurrent transfers
/// the scheduler has ~3N runnable threads and thirty-two places to put them,
/// while only eight of them may run at a time. Every hand-off then crosses a
/// core, and a hand-off that crosses a core is a futex.
///
/// Measured on the reference desktop, 2026-08-04, one binary, one 8-CPU budget, the `year`
/// corpus (28 893 objects) over smart HTTP, ramp 1→32 clients:
///
/// | budget applied as | CPU-seconds per request, 1 client → 32 |
/// |---|---|
/// | `cpu.max` quota only, 32 CPUs allowed | 0.42 → 0.94 |
/// | the same quota, affinity pinned to 8 | 0.34 → 0.35 |
///
/// The quota-only arm spent **40% of the server's CPU** in the kernel's futex
/// hash-bucket spinlock (`native_queued_spin_lock_slowpath`, `perf record` on
/// the serving process). The pinned arm is flat: gunnar's per-request cost does
/// not in fact grow with concurrency, and the ramp that said it did was
/// measuring where the threads were allowed to run.
///
/// So this is a line in the log and not a change of behaviour: gunnar must not
/// pin itself. It does not know which CPUs are *its* to take, and choosing wrong
/// would be worse than the condition being reported. The operator does know, and
/// `--cpuset-cpus` is the whole fix.
///
/// The decision is [`spread_budget`], separately, because a condition read off
/// the host it is running on can only ever be seen in the state that host
/// happens to be in — and this one has to be seen in both.
fn report_cpu_budget() {
let Some((allowed, budget)) = spread_budget(cpus_allowed(), cpu_quota()) else {
return;
};
tracing::warn!(
cpus_allowed = allowed,
cpu_quota = budget,
"this server may run on {allowed} CPUs but may only use {budget} CPUs' worth of \
them. Every hand-off between the three threads that serve one transfer then \
crosses a core, and in a 2026-08-04 measurement that cost 2.7x the CPU per \
request at 32 concurrent clients against the identical budget applied as a \
cpuset. Start the \
container with `--cpuset-cpus` (or the unit with `AllowedCPUs=`) naming {budget} \
CPUs, not only `--cpus`/`CPUQuota=`."
);
}
/// The runtime `gunnar serve` runs the **git** listener on.
///
/// Built here rather than with `#[tokio::main]` so a client-only build links no
/// async runtime at all, and built with an explicit
/// [`GIT_BLOCKING_THREADS`] so the number that used to be the server's
/// accidental concurrency ceiling is now written down somewhere.
pub fn runtime() -> Result<tokio::runtime::Runtime> {
tokio::runtime::Builder::new_multi_thread()
.max_blocking_threads(GIT_BLOCKING_THREADS)
.thread_name("gunnar-git")
.enable_all()
.build()
.context("building the git runtime")
}
/// Both listeners, bound and running.
/// How the smart-HTTP listener gets its TLS identity (see
/// [`ServeConfig::http_tls`]).
#[cfg(feature = "http")]
#[derive(Debug, Clone)]
pub enum HttpTls {
/// Mint a leaf from this server's own issuer at start — the
/// single-executable deployment. Refuses to start if it cannot mint;
/// silent cleartext fallback is the defect this enum exists to prevent.
SelfMinted,
/// Operator-supplied chain and key.
Identity(gunnar_http::TlsIdentity),
}
/// How the gRPC control plane gets its **mutual** TLS material.
///
/// The mirror of [`HttpTls`], with one difference that is the whole point: the
/// smart-HTTP listener terminates TLS to encrypt, and this one terminates TLS
/// to *identify*. So there is no bring-your-own-CA arm here yet. A certificate
/// the gate can turn into a principal has to have been registered in the ACL
/// table, and the only thing that mints and registers in one act is this
/// server's own issuer (`Bootstrap.Enroll`, `gunnar control bootstrap`).
/// Trusting a foreign CA would mean a certificate that passes the handshake and
/// can never pass the gate, which is a confusing way to be refused.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GrpcTls {
/// Mint the listener's own leaf from this server's issuer at start, and
/// pin that issuer as the CA every client certificate must come from.
///
/// Refuses to start if it cannot mint. A control plane that asked for
/// client certificates and came up without them would be the same silent
/// downgrade [`HttpTls`] refuses.
SelfMinted,
}
/// **A tokio `Runtime` that is always torn down with `shutdown_background`,
/// never by a blocking `Drop`.**
///
/// Dropping a `Runtime` blocks the current thread on its worker and blocking
/// pools' shutdown, and tokio **panics** when that blocking drop happens inside
/// another runtime's context: *"Cannot drop a runtime in a context where
/// blocking is not allowed."*
///
/// [`Server::start`] is an `async fn` that builds the control plane's own
/// runtime and then runs several more fallible steps before it moves that
/// runtime into [`Server`]. A `?` anywhere in that window used to drop a bare
/// `Runtime` on the async stack — and on the appliance gunnar is **PID 1**, so
/// that panic is the kernel's `Attempted to kill init`. Measured 2026-08-31:
/// a power cut left a 0-byte `replication.json`, its parse in `start`
/// failed, the `?` unwound past the live runtime, and the box wedged on every
/// reboot. Wrapping the runtime in this guard makes the early return safe **by
/// construction** — one drop path, and it never blocks — instead of by
/// remembering to hand-`take()` it at each of the several exits.
///
/// Since R5.1-G it holds the control plane's TWO runtimes — the connections'
/// and the calls' — under the one guard, in that order.
#[derive(Default)]
struct ControlRuntime(Vec<tokio::runtime::Runtime>);
impl ControlRuntime {
/// Hold a runtime whose drop must be non-blocking.
fn new(runtime: tokio::runtime::Runtime) -> Self {
ControlRuntime(vec![runtime])
}
/// Take over every runtime `other` holds, so one guard drops them all.
fn hold(&mut self, other: &mut ControlRuntime) {
self.0.append(&mut other.0);
}
/// A handle on the first runtime held, if any.
fn handle(&self) -> Option<tokio::runtime::Handle> {
self.0.first().map(|runtime| runtime.handle().clone())
}
/// Tear the runtimes down now, without blocking, if any are still held.
/// Idempotent, so an explicit call and the `Drop` cannot double-free.
fn shutdown(&mut self) {
for runtime in self.0.drain(..) {
runtime.shutdown_background();
}
}
}
impl Drop for ControlRuntime {
fn drop(&mut self) {
self.shutdown();
}
}
pub struct Server {
/// An `Option` only because [`Server`] now has a [`Drop`] impl, which makes
/// moving a field out of `self` in [`Server::shutdown`] illegal.
ssh: Option<BoundServer>,
grpc: Option<JoinHandle<Result<()>>>,
grpc_shutdown: Option<oneshot::Sender<()>>,
/// The bound gRPC address and whether that listener authenticates, as
/// ONE field. They are printed together and must not be able to disagree.
grpc_endpoint: Option<(SocketAddr, bool)>,
/// The control plane's **own** runtime, kept alive here because
/// dropping it would stop the listener it carries.
///
/// A [`ControlRuntime`] guard, not a bare `Runtime`: dropping a `Runtime`
/// blocks, and doing so from inside another runtime's context panics
/// ("Cannot drop a runtime in a context where blocking is not allowed"),
/// and both [`Server::start`] and [`Server::shutdown`] run in async
/// context. The guard routes *every* drop through `shutdown_background`, so
/// no `?`, panic or forgotten `take()` can turn a teardown into an abort —
/// which as PID 1 is a kernel panic.
control_runtime: ControlRuntime,
stores: Arc<ServerStores>,
/// Whether the unauthenticated `Bootstrap.Enroll` window was open when this
/// server started, so the banner on stdout says the same thing the log does.
///
/// A field rather than a re-read: the banner and the warning must not be
/// able to disagree about the single most important fact a fresh server has
/// (one implementation).
bootstrap_window_open: bool,
/// **The SSH host key this server is serving, `SHA256:…`.**
///
/// Kept so the startup banner can print it. It is logged at start through
/// `tracing`, which on the appliance means stderr, which means
/// `/dev/console`, which is the LAST `console=` on the command line, which
/// is the serial line — and the UpCloud shape has no serial line. So the
/// one emission this box makes of the one fact a client has to pin
/// reached nobody in production, and `known_hosts` got filled by scanning
/// instead. A field, not a re-read, for the same reason
/// `bootstrap_window_open` is one: the banner and the log must not be able
/// to disagree about it.
host_key_fingerprint: String,
/// The counters both listeners write to.
metrics: Arc<ServerMetrics>,
/// The heartbeat task, aborted at shutdown so a stopped server stops
/// narrating.
heartbeat: Option<JoinHandle<()>>,
/// I11's standby watch, held so it is aborted with the server. A watchdog
/// that outlived the box it watches over could promote a server that is
/// shutting down.
#[cfg(feature = "replication")]
standby_watch: Option<JoinHandle<()>>,
/// The boot check and the pull channel ([`crate::repair`]), aborted with the
/// server (a recharge it was running fails its restore stream, never ends it).
#[cfg(feature = "replication")]
repair: Option<JoinHandle<()>>,
/// The event bus both the transport and the control plane publish to.
bus: Arc<EventBus>,
/// The smart-HTTP listener's task and the address it bound (smart HTTP).
/// Dropped with the `Server`, and therefore aborted.
#[cfg(feature = "http")]
http: Option<(JoinHandle<std::io::Result<()>>, SocketAddr, bool)>,
/// The scrape listener's task and the address it bound. Dropped
/// with the `Server`, and therefore aborted: a stopped server must stop
/// answering, or a scraper keeps drawing the last numbers it saw.
scrape: Option<(JoinHandle<std::io::Result<()>>, SocketAddr)>,
/// The replicator, when any standby is configured.
/// the replicator and its workers, in the LIVE SLOT the control plane
/// reconfigures through.
///
/// It was an `Option<Arc<Replicator>>` plus an `Option<ReplicationTasks>`
/// until 2026-08-30 — decided once at startup, from a file. That is
/// unusable on the shape this ships to: an OS-less appliance with no shell,
/// where the only way to name a standby is over the control plane. See
/// [`ReplicationSlot`](crate::mirror::ReplicationSlot).
///
/// Dropped with the `Server`, and therefore aborted, so a stopped server
/// stops replicating.
#[cfg(feature = "replication")]
replication: Arc<crate::mirror::ReplicationSlot>,
/// The outbound webhook delivery slot, reconfigured through the control plane.
/// Dropped with the `Server`, and therefore aborted, so a stopped server
/// stops delivering.
#[cfg(feature = "webhooks")]
webhook: Arc<crate::webhook::WebhookSlot>,
}
/// One line every `interval` naming what the server has done, but only when
/// something moved.
///
/// **Not a scrape endpoint.** The load target is a forge, and the first
/// question at an incident gets answered from the log the operator already has
/// open. A Prometheus surface is a separate decision with its own cost: another
/// listener, and another thing to authenticate. This makes the counters
/// readable today, in four lines.
async fn heartbeat(metrics: Arc<ServerMetrics>, interval: std::time::Duration) {
let mut ticker = tokio::time::interval(interval);
// The first tick fires immediately; skip it so start-up does not emit a row
// of zeroes that reads like traffic.
ticker.tick().await;
let mut previous = metrics.snapshot();
loop {
ticker.tick().await;
let now = metrics.snapshot();
// An idle server says nothing. A server that has served ANYTHING since
// the last beat says what, including one whose only change is that a
// transfer finished; that is how `transfers_in_flight` coming back to
// zero becomes visible.
if now == previous {
continue;
}
emit_metrics(&now);
previous = now;
}
}
/// The gRPC control plane's request ids, minted from the **same** counter the
/// SSH listener and the smart-HTTP listener mint theirs from.
///
/// This one line is why `gunnar_grpc::router` takes a source instead of owning
/// one: `gunnar-grpc` has no `gunnar-core` edge and must not grow one (it would
/// put `aws-lc-rs`, C and a build script into the `--features client,control`
/// graph), and a private counter over there would mean a control-plane call and
/// a git transfer could print the same `req=` value in one log.
///
/// `gunnar_grpc::trace`'s rendering is asserted equal to
/// [`RequestId`]'s by `the_grpc_request_id_renders_exactly_as_gunnar_core_
/// renders_one` in `tests/observability.rs`, because a shared counter rendered
/// two ways is still two identifiers.
fn grpc_request_id() -> u64 {
RequestId::next().get()
}
/// The heartbeat line.
///
/// **This function used to BE the line**, and it listed eighteen of the
/// nineteen counters. `connections_anonymous` — written by both transports
/// (`gunnar-ssh/src/server.rs`, `gunnar-http/src/service.rs`), and the answer to
/// *how much of my traffic is strangers*, which the field's own documentation
/// calls the first question an operator asks after making anything public —
/// appeared in no line gunnar had ever printed. Nothing was wrong with the
/// counter; the hand-written list next to it had simply never been updated when
/// the field was added.
///
/// So it is not a hand-written list any more. The line expands from the same
/// macro arm that declares the fields, in `gunnar-core`, beside the scrape
/// endpoint's renderer, and neither can fall behind the struct (one implementation). This
/// wrapper stays only so the heartbeat task and a test call the same one thing.
fn emit_metrics(now: &MetricsSnapshot) {
now.emit_heartbeat();
}
impl Server {
/// Bind everything and start accepting.
///
/// Returns once both sockets are bound, so a caller knows the real
/// addresses before the first client connects, which is what makes port 0
/// usable in a test.
pub async fn start(stores: Arc<ServerStores>, config: ServeConfig) -> Result<Server> {
report_cpu_budget();
// The bind spelling becomes an address HERE, before anything else in
// this function can report one. `run` has already resolved it and
// replaced the spec with the answer, so on the appliance path this is
// the identity case; a caller that reaches `start` directly (every
// test, and `gunnar serve` under an init system) gets the resolution
// it would otherwise have had to do itself.
let mut config = config;
// A bind by ROLE (`utility:50051`, the lean estate) becomes this box's NIC first.
config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)?;
// The interface spelling names a DEVICE as well as an address; keep the
// device before the spec becomes an address, so the listener can bind it.
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
config.grpc_bind_device = Some(interface.clone());
}
// **An interface whose lease PID 1 still asks for** (its boot DHCP failed —
// estate-box3 D4, a box rebooted while its network was down) is WAITED for: the
// control plane binds once the address arrives instead of never.
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
let has_address =
|i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
}
// ★ **An estate box is WireGuard-only** (owner 2026-10-04: every
// box-to-box connection over the tunnel), whatever its medium says: a
// pairing seed with a WireGuard tunnel IS an estate (the front, the
// relay, the overlay). A standalone appliance — no seed — is not.
#[cfg(feature = "wg-client")]
if !config.wg_only {
if let Ok(Some(seed)) = crate::pairing::seed(stores.root()) {
if crate::tunnel_addr::estate_is_wg_only(Some(
seed.tunnel == nordisk_pairing::seed::TunnelKind::Wireguard,
)) {
config.wg_only = true;
tracing::warn!("serve: this box belongs to an estate (a WireGuard pairing seed) — WireGuard-only: its utility address serves nothing, boxes speak over the tunnel");
}
}
}
// The posture the pairing reads (which address of the primary a twin joins).
crate::tunnel_addr::set_wg_only(config.wg_only);
let resolved_grpc_bind = match resolve_grpc_bind(&config.grpc_bind)? {
Some(addr) => addr,
None => {
// The interface is there and holds no address. There is nothing
// to bind, so there is no control plane — the same shape
// `--no-grpc` produces, reached for the same honest reason and
// reported through the same field, so the dashboard row, the
// readiness line and `Admin.Health` all agree.
config.no_grpc = true;
gunnar_grpc::DEFAULT_BIND
}
};
// ── the bootstrap: refuse to serve rather than start unreachable ───
let admin_key = match &config.admin_key {
Some(spec) => Some(read_key_argument(spec)?),
None => None,
};
// The keys a restore is taken from, resolved and CHECKED now: a line that does not
// parse would otherwise be a key that silently lets nobody in, found out on the day
// of the restore.
let restore_keys = config
.restore_pubkeys
.iter()
.map(|spec| {
let line = read_key_argument(spec)?;
gunnar_core::auth::parse_public_key(line.trim()).map_err(|err| {
anyhow::anyhow!(
"--restore-pubkey {spec:?} is not an OpenSSH public key line: {err:#}"
)
})?;
Ok(line.trim().to_owned())
})
.collect::<Result<Vec<String>>>()?;
let readiness = stores
.auth()
.prepare_to_serve(admin_key.as_deref())
.with_context(|| {
format!(
"reading the identity table at {}. This is NOT the same as finding it \
empty: a data path that cannot be read must never be served as a server \
nobody has enrolled yet, because that would reopen the bootstrap window on \
a disk fault.",
stores.root().display()
)
})?;
announce_readiness(
&readiness,
stores.root(),
resolved_grpc_bind,
config.no_grpc,
);
let bootstrap_window_open = readiness.bootstrap_window_open();
// ── the SSH host key, minted in-process and persisted ──────────────
// A twin frames nothing: its host key is read (the primary's, replayed)
// or minted through the UNLOGGED store, and its follower starts here.
let host_key_store = if stores.is_twin() {
stores.secrets_unframed()?
} else {
stores.secrets()?
};
let host_key =
ensure_host_key(&host_key_store).context("loading or minting the SSH host key")?;
if stores.is_twin() {
crate::twin::start_follower(Arc::clone(&stores));
}
// The pairing loop, once the control plane serves: join the front,
// take the role, become the twin if elected (pairing.rs).
#[cfg(feature = "wg-client")]
match crate::pairing::seed(stores.root()) {
Ok(Some(seed)) => crate::pairing::start(Arc::clone(&stores), seed),
Ok(None) => {}
Err(e) => {
tracing::error!(error = %format!("{e:#}"), "pairing: the seed is present and unusable — this box will not pair")
}
}
let host_key_fingerprint = host_key.fingerprint().to_string();
tracing::info!(
fingerprint = %host_key_fingerprint,
"ssh host key ready (persisted; it survives a restart)"
);
// The fingerprint crosses the ssh-key version split as BYTES: gunnar-core
// links ssh-key 0.6 and russh pins 0.7.0-rc.10, so no key type can be
// handed from one to the other.
let russh_key = russh_private_key(host_key.private_openssh())
.context("handing the host key to russh")?;
// ── the certificate issuer, minted at first start ──────────────────
//
// The same discipline as the host key above: a fresh appliance mints
// its own, and a restart finds the one it minted. `Pki::create`'s
// refusal lives under the document's compare-and-swap, so two racing
// starts cannot mint two issuers. Failure is a warning, not a refusal
// to serve: SSH-key auth carries the server whether or not certificates
// exist, and an operator who imports a CA later gets exactly the
// `issuer import` refusal they should.
match stores.pki() {
Ok(pki) => match pki.issuer() {
Ok(Some(live)) => tracing::info!(
subject = %live.subject,
fingerprint = %live.fingerprint,
"issuer ready (persisted; it survives a restart)"
),
Ok(None) => match pki.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS) {
Ok(created) => tracing::warn!(
subject = %created.subject,
fingerprint = %created.fingerprint,
"FIRST START: this server minted its own certificate issuer. Its \
private key is sealed in the secret store. `gunnar issuer show --pem` \
prints the certificate clients pin."
),
Err(err) => tracing::warn!(
%err,
"the issuer could not be minted at first start; certificate issuance \
is unavailable until `gunnar issuer create` or `issuer import` succeeds"
),
},
Err(err) => tracing::warn!(%err, "the PKI document could not be read"),
},
Err(err) => tracing::warn!(
%err,
"no secret store, so no issuer: certificate issuance is unavailable"
),
}
// ── can this server be moved onto mutual TLS? ──────────────────────
//
// Asked HERE, and the position is two decisions rather than one.
//
// **After the issuer block above**, because that block MINTS one on a
// first start: asking before it would refuse a server whose issuer is
// four lines from existing, and the refusal would name the one thing
// the operator did not have to do.
//
// **Before anything binds, and before the control plane's runtime is
// built.** Two reasons:
//
// 1. `--grpc-tls-self` is a one-way door for everybody who cannot
// complete the handshake afterwards, and exactly one state makes
// it a door with nobody on either side. A server asked for client
// certificates that cannot honour the ask must be a startup error
// naming the cause, never a listener that came up — the same
// discipline `--http-tls-*` already follows.
// 2. **A `?` past the control runtime is an appliance that will not
// boot.** `ControlRuntime`'s own docs record the measurement: that
// `Runtime` is built bare inside this function and dropping one on
// an async stack panics, and on the appliance gunnar is PID 1, so
// the panic is `Attempted to kill init`. The first draft of this
// check sat beside the trust root, inside that window, and
// `tests/tls_transition.rs` failed with exactly that panic instead
// of the refusal it was asserting. That test is the guard that it
// does not move back.
//
// `crate::tls_preflight` holds the argument and the verdicts;
// `gunnar issuer readiness` is this same function asked in advance,
// so the answer an operator plans against is the answer acted on here
// (one implementation).
if !config.no_grpc && config.grpc_tls.is_some() {
let preflight = crate::tls_preflight::preflight(&stores)
.context("checking whether this server can move to mutual TLS")?;
if let Some(refusal) = preflight.refusal() {
anyhow::bail!("{refusal}");
}
preflight.warn();
}
// ── the SSH listener ───────────────────────────────────────────────
//
// ONE set of counters, minted here and shared by the listener
// and the transport. Two `ServerMetrics` would produce two partial
// truths, the listener's refusals in one and the transport's pushes in
// the other, and neither answers "what is this server doing".
// The stores' set (A28 M17): the flow flush's reconcile pass, started before this,
// counts its misses on it too.
let metrics = Arc::clone(stores.metrics());
let bus = Arc::new(EventBus::new(gunnar_grpc::DEFAULT_EVENT_BUFFER));
let transport = Arc::new(
WireTransport::new(Arc::clone(&stores))
.with_bus(Arc::clone(&bus))
.with_metrics(Arc::clone(&metrics))
.with_wire_config(config.wire.clone()),
);
let defaults = SshConfig::default();
// The DECLARED budget (`crate::budget`): cores x TRANSFERS_PER_CORE,
// capped by the descriptors the handle cache has not already spoken
// for, with `MemAvailable` sizing the ledger. An operator's
// `--max-concurrent-transfers` still wins — `.or()` below — and so does
// `GUNNAR_MAX_TRANSFERS`, which is how the appliance's kernel command
// line reaches it. Before this the ceiling was the shipped 256, which
// the governor then walked down to 19 under load while the
// ledger's flat 320 MiB-per-push refused everything past 25.
let mut budgets = crate::budget::budgets();
// **Recorded, so the box can say later whether its socket is where a
// twin's belongs** (2026-09-21, lane T4). A box told it is a twin AFTER
// this line keeps whatever was bound here until it restarts;
// `twin::git_listener_needs_restart` is how that is reported instead of
// being left to `ss`.
let twin_shaped = stores.is_twin() || config.twin_role;
crate::twin::note_git_bind_is_twin_shaped(twin_shaped);
let ssh_config = SshConfig {
bind: if twin_shaped {
// **No forge listener on a twin.** It serves no git to anyone: a
// read would grow the caches the twin must not hold. The listener
// the rest of `start` is wired around stays, on loopback, port 0.
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, asked = %config.ssh_bind, "this box is a TWIN: the git listener is bound to loopback only");
SocketAddr::from(([127, 0, 0, 1], 0))
} else {
config.ssh_bind
},
allow_anonymous: config.allow_anonymous,
// The banner names the BUILD, not just the crate version. On the
// OS-less appliance this listener is the only thing that can be
// asked what is running — there is no shell to run `gunnar
// --version` in. See `gunnar_common::ssh_server_id`.
server_id: gunnar_common::ssh_server_id(),
// `or` rather than `unwrap_or`: `None` here means "the operator did
// not say", which takes the default. Turning a limit *off* is
// `Some(0)`-free and stays a code path, not a flag, because an
// appliance with no transfer ceiling is the unbounded listener
// exactly as it was.
max_concurrent_transfers: config.max_concurrent_transfers.or(Some(budgets.transfers)),
memory_budget_bytes: config.memory_budget_bytes.or(Some(budgets.memory_bytes)),
admission_wait: config.admission_wait.or(defaults.admission_wait),
transfer_idle_timeout: match config.transfer_idle {
None => defaults.transfer_idle_timeout,
Some(d) if d.is_zero() => None,
Some(d) => Some(d),
},
max_transfers_per_principal: config
.max_transfers_per_principal
.or(defaults.max_transfers_per_principal),
// `or` for the same reason as the two above: `None` means the
// operator did not say, which takes `SshConfig::default`'s
// ceiling. Switching it OFF is deliberately not reachable from a
// flag — `Some(0)` refuses everything anonymous, and letting
// anonymous inherit a keyed identity's share is the hole anonymous read opened
// and this closed, not a configuration.
max_anonymous_transfers: config
.max_anonymous_transfers
.or(defaults.max_anonymous_transfers),
per_principal_limit_is_hard: config.per_principal_limit_is_hard,
#[cfg(feature = "admit-share")]
admit: config.admit,
..defaults
};
// The row says what is IN FORCE, after the operator's flags: a line
// that printed the derivation while the server ran on something else
// would be worse than no line, because the first thing anybody does
// with it is compare it to the refusals underneath it.
if let Some(n) = config.max_concurrent_transfers {
budgets.transfers = n;
budgets.transfers_from = crate::budget::Source::Operator;
}
if let Some(bytes) = config.memory_budget_bytes {
budgets.memory_bytes = bytes;
}
if let Some(wait) = config.admission_wait {
budgets.wait_ms = wait.as_millis() as u64;
}
crate::budget::announce(&budgets);
#[cfg(feature = "http")]
let http_transport: Arc<dyn gunnar_ssh::seam::GitTransport> = Arc::clone(&transport) as _;
// The SAME `WireTransport`, reached through a second trait rather than
// a second object: `open_lfs` hands out the store under the repository's
// own git dir and the `WireConfig::lfs` limits the SSH session is built
// with, so an operator's `--max-lfs-object-bytes` binds on both
// transports or on neither.
// ── the metered appliance's entitlement slot, before any front door ─
//
// Opened here so BOTH front doors and the control plane share ONE
// slot (one implementation): the same arc-swap the SSH dispatch reads is the one
// `Entitlement.Set` swaps. Loading it re-verifies every stored fact
// (see `crate::entitlement`), and a key that does not parse is a
// startup error — a metered box that silently ran Free would be the
// provider's revenue leaking with no line to say so.
#[cfg(feature = "metered")]
let entitlement = match &config.metered {
Some(metered) => Some(
crate::entitlement::EntitlementSlot::open(Arc::clone(&stores), metered)
.context("opening the entitlement slot")?,
),
None => {
tracing::warn!(
"metered build started without --monetize-pubkey: every namespace is Free \
and Entitlement.Set refuses every fact"
);
None
}
};
// The band's tenants row reads the SAME slot `Admin.Stats` counts from
// (one implementation). Filled here, before any door and outside the gRPC branch, so
// a `--no-grpc` metered box still shows its tenants.
// **This estate SELLS room** (the feed's `sells_room`, LIVE; gunnar-grpc `selling`):
// ONE flag, read by the entitlement slot's push gate and the gRPC context alike.
#[cfg(feature = "metered")]
let sells_room = entitlement
.as_ref()
.map(|slot| slot.sells_room_flag())
.unwrap_or_else(|| Arc::new(std::sync::atomic::AtomicBool::new(false)));
#[cfg(not(feature = "metered"))]
let sells_room = Arc::new(std::sync::atomic::AtomicBool::new(false));
// **The front's egress door** (the lean estate, A19): a box laid with no public IPv4 is
// fed it, and every client of this box that reaches the internet takes it from then on.
nordisk_pairing::feedbus::on_published("the front's egress door", crate::feed::take_egress);
{
let flag = Arc::clone(&sells_room);
nordisk_pairing::feedbus::on_published(
"shop room: a person's account with no fact",
move |feed| {
let sells_room = crate::feed::Fed::from_feed(feed).sells_room;
if flag.swap(sells_room, std::sync::atomic::Ordering::Relaxed) != sells_room {
tracing::info!(target: "gunnar::audit", event = "shop_room", sells_room, "{}", if sells_room { "this estate's room comes from the shop: a person's own account with no fact holds no room until its fact lands" } else { "this estate's room does not come from a shop: an account with no fact is uncapped" });
}
},
);
}
#[cfg(feature = "metered")]
if let Some(slot) = &entitlement {
// A key the installer feeds later is applied to THIS slot in place
// (owner 2026-09-25: avoid restarts; `feed::monetize_arrived`).
let taker = Arc::clone(slot);
crate::feed::on_monetize_key(
"entitlement slot",
Box::new(move |k| taker.set_key(k).map(|_| ())),
);
// ★ UW-6: the SITE's pay page, fed per estate (`refusal_link`), replaces the
// constant medium's — now and on every feed after (LIVE).
let pay = Arc::clone(slot);
nordisk_pairing::feedbus::on_published(
"entitlement: the site's pay page",
move |feed| {
if let Some(url) = crate::feed::Fed::from_feed(feed).billing_url.as_deref() {
if let Err(why) = pay.set_billing_url(url) {
tracing::warn!(target: "gunnar::audit", event = "entitlement.billing_url_refused", reason = %why, "the fed pay page was refused; the one in force stays");
}
}
},
);
let slot = Arc::clone(slot);
config.tenants_view.fill(move || slot.counts());
}
// ── spools a crash left behind (cov-13) ──
//
// A pushed pack of 64 MiB or more is spooled to `incoming-*.pack` while
// it arrives and removed by `Drop` when it lands or fails. A process
// that died mid-receive ran no `Drop`, so the file is still here, full
// size, under a name nothing reads — room the next push is refused for.
// No door is open yet, so no live push owns one.
let (mut swept, mut swept_bytes) =
crate::push_space::sweep_spool_leftovers(stores.repos_root());
// …and an LFS upload's `<txlog>/spool/*.part`, the same shape one door over.
let (lfs, lfs_bytes) = crate::push_space::sweep_lfs_spool(
&stores.txlog().dir().join(gunnar_core::txlog::SPOOL_DIR),
);
swept += lfs;
swept_bytes += lfs_bytes;
if swept > 0 {
tracing::warn!(
files = swept,
bytes = swept_bytes,
"deleted {swept} pushed-pack spool file(s) a crash left behind ({})",
gunnar_ssh::headroom::human(swept_bytes)
);
}
// ── the disk-headroom gate: the community plan gate, every build ──
//
// Measured once here so the boot log states the floor IN FORCE beside
// what is free right now; refused per request in the chain below and
// in `Repos.Create`. A root that cannot be measured is logged as the
// refusal it will be, not hidden until the first push finds it.
//
// ★ The filesystem's preallocation, measured once here (round 5.2): what a growing
// spool-sized file occupies beyond its bytes on THIS data set (XFS doubles it; tmpfs and
// ext4 add nothing), so every push reserves what it will really hold while it lands.
let (fs_slack, unmeasured) = gunnar_ssh::space::FsSlack::of(stores.root());
match &unmeasured {
None => {
tracing::info!(root = %stores.root().display(), measured_permille = ?fs_slack.measured_permille, slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), None))
}
Some(why) => {
tracing::warn!(root = %stores.root().display(), slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), Some(why)))
}
}
let ledger = match &config.space_probe {
None => gunnar_ssh::SpaceLedger::new(stores.root(), config.disk_floor),
Some(probe) => gunnar_ssh::SpaceLedger::with_probe(
stores.root(),
config.disk_floor,
Arc::clone(probe),
),
};
let disk_gate = Arc::new(gunnar_ssh::DiskHeadroomGate::over(Arc::new(
ledger.with_fs_slack(fs_slack),
)));
match disk_gate.headroom() {
Ok(room) => tracing::info!(
root = %stores.root().display(),
free = %gunnar_ssh::headroom::human(room.free),
floor = %gunnar_ssh::headroom::human(room.floor),
disk_floor = %config.disk_floor,
has_room = room.has_room(),
"disk headroom: pushes and Repos.Create are refused under the floor"
),
Err(err) => tracing::error!(
root = %stores.root().display(),
%err,
"disk headroom could not be measured; every push and Repos.Create will be \
refused until it can be"
),
}
// The front doors' control: the gate CHAIN, disk first, then the plan.
let front_control = || {
front_control(
&stores,
config.anonymous.clone(),
Arc::clone(&disk_gate),
#[cfg(feature = "metered")]
entitlement.as_ref(),
)
};
#[cfg(feature = "http")]
let http_lfs: Arc<dyn gunnar_ssh::seam::LfsTransport> = Arc::clone(&transport) as _;
// **Repository settings over the forge's door** (owner, 2026-10-14):
// `gunnar-repo …` on this listener, answered by the control plane's own
// `Repos` handlers once the control plane exists (set further down);
// until then — or on `--no-grpc` — the verb says it is not answered here.
let repo_door_ctx: Arc<std::sync::OnceLock<gunnar_grpc::service::ServiceContext>> =
Arc::new(std::sync::OnceLock::new());
let ssh = GunnarSsh::new(ssh_config, front_control(), transport)
// The boot window (`crate::readiness`): while the start-up disk
// test is saturating /data, a git transfer is refused with
// `starting` and EX_TEMPFAIL rather than served an open that
// fails and reads as a missing repository.
.with_readiness(crate::readiness::flag())
// The admission gate is the ONE door all three git fronts ask, so
// it is where a refusal is charged to the tenant that met it. See
// `GunnarSsh::with_flow`.
.with_flow(Arc::clone(stores.flow()))
.with_metrics(Arc::clone(&metrics));
// **The sealed box's tunnel-address door** (T17): `gunnar-tunnel-address`
// over this listener, server-wide Admin, answered by the SAME function
// `Bootstrap.AssignTunnelAddress` and `Admin.SetTunnelAddress` call. The
// listener already held the caller to server Admin, so `may_write`.
#[cfg(feature = "wg-client")]
let ssh = {
let root = stores.root().to_path_buf();
ssh.with_tunnel_address_door(Arc::new(move |address: &str| {
crate::appliance::tunnel_address_answer(&root, address, true)
.map(|answer| crate::appliance::render_tunnel_answer(&answer))
.map_err(|err| format!("{err}"))
}))
};
let ssh = {
let cell = Arc::clone(&repo_door_ctx);
ssh.with_repo_door(Arc::new(move |ask: gunnar_ssh::RepoAsk| {
let cell = Arc::clone(&cell);
Box::pin(async move {
let Some(ctx) = cell.get() else {
return Err(
"this server answers repository settings only on its control plane"
.to_owned(),
);
};
let setting = match ask.action {
gunnar_ssh::RepoAction::Show => gunnar_grpc::repo_door::RepoSetting::Show,
gunnar_ssh::RepoAction::Visibility(public) => {
gunnar_grpc::repo_door::RepoSetting::Visibility(public)
}
gunnar_ssh::RepoAction::DefaultBranch(b) => {
gunnar_grpc::repo_door::RepoSetting::DefaultBranch(b)
}
gunnar_ssh::RepoAction::Create { sha256 } => {
gunnar_grpc::repo_door::RepoSetting::Create(sha256)
}
};
gunnar_grpc::repo_door::ask(
ctx,
&ask.principal,
&ask.fingerprint,
ask.repo.namespace(),
ask.repo.name(),
setting,
)
.await
.map(|repo| gunnar_grpc::repo_door::render(&repo))
.map_err(|status| status.message().to_owned())
})
}))
};
// **--wg-only** (owner 2026-10-04): a connection at the utility address
// — where the control plane is bound — is refused before a byte; the
// public's git clients and the tunnel are served as ever.
let ssh = if config.wg_only && !config.no_grpc {
let utility = resolved_grpc_bind.ip();
tracing::warn!(%utility, "ssh: --wg-only — a connection at the utility address is refused; boxes speak over the tunnel");
ssh.with_refused_local(Arc::new(move |local: std::net::IpAddr| local == utility))
} else {
ssh
};
let ssh = ssh
.bind(russh_key)
.await
.with_context(|| format!("binding the SSH listener on {}", config.ssh_bind))?;
// The anonymous setting is on the *bind* line rather than in a separate
// one, so an operator reading a boot log sees what this listener will
// serve strangers in the same place they see where it is listening.
tracing::info!(
addr = %ssh.local_addr(),
anonymous_read = config.allow_anonymous,
"gunnar ssh listener: git-upload-pack, git-receive-pack; publickey, plus \
anonymous read of public repositories when enabled"
);
// ── the smart-HTTP listener ────────────────────────────────────────
//
// The SAME `Dispatch` the SSH listener authorises with, the SAME
// `WireTransport`, the SAME counters. That is the whole point of the
// crate: a second transport, not a second server. If this ever needed
// its own control plane or its own transport, HTTP and SSH would be
// able to disagree about who may push.
#[cfg(feature = "http")]
let http = match config.http_bind {
None => None,
Some(bind) => {
// The TLS identity, resolved BEFORE binding: a `--http-tls-*`
// that cannot produce one must be "the server refused to
// start", never a listener that quietly came up cleartext —
// that silent fallback would be the whole defect the flag
// exists to prevent.
let tls = match &config.http_tls {
None => None,
Some(HttpTls::Identity(identity)) => Some(identity.clone()),
Some(HttpTls::SelfMinted) => {
let pki = stores.pki().context(
"--http-tls-self needs the secret store; no PKI is available",
)?;
let issuer = pki
.issuer()
.context("reading the issuer for --http-tls-self")?
.context(
"--http-tls-self needs an issuer and none exists; the first-start \
mint above must have failed — its warning says why",
)?;
// The names a client may dial this listener by. rcgen
// types each SAN as IP or DNS on its own. Loopback is
// always present so `git clone https://127.0.0.1/…`
// works out of the box; a concrete bind address joins
// it. An operator with real DNS names brings their own
// certificate — that is what `--http-tls-cert` is for.
let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
let ip = bind.ip();
if !ip.is_unspecified() && !ip.is_loopback() {
hosts.push(ip.to_string());
}
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the smart-HTTP server certificate")?;
let key_pem = issued.key_pem.clone().context(
"the issuer minted a server certificate but returned no key",
)?;
tracing::info!(
fingerprint = %issued.fingerprint,
sans = ?hosts,
"smart-HTTP TLS: server certificate minted from this server's own \
issuer; clients pin the issuer (`gunnar issuer show --pem`)"
);
Some(gunnar_http::TlsIdentity {
// Leaf first, issuer after, so a pinning client can
// build the chain from the handshake alone.
cert_chain_pem: format!("{}{}", issued.cert_pem, issuer.cert_pem),
key_pem,
})
}
};
let http_config = gunnar_http::HttpConfig {
listen: bind,
// One field, both listeners. See `ServeConfig::allow_anonymous`.
allow_anonymous: config.allow_anonymous,
cleartext: config.http_cleartext,
tls,
..gunnar_http::HttpConfig::default()
};
let bound = gunnar_http::GunnarHttp::new(
http_config,
gunnar_ssh::server::Dispatch::new(front_control()),
Arc::clone(&http_transport),
)
.with_metrics(Arc::clone(&metrics))
// ── ONE admission gate for BOTH front doors ─────────────────
//
// The SSH listener's own gate, handed in rather than rebuilt,
// so a git transfer over smart HTTP counts against the same
// server-wide limit, the same per-principal share and the same
// memory budget as one over SSH. Before this the HTTP front had
// no gate at all (measured: 64 clients, 64 slots, nothing
// refused); two independent gates would have been the next
// worst thing, doubling the appliance's real ceiling in
// silence. This is the same-writer fix the crate's own comment
// above demands: "a second transport, not a second server."
.with_admission(ssh.admission_gate())
.with_lfs(Arc::clone(&http_lfs))
.bind()
.await
.with_context(|| format!("binding the HTTP listener on {bind}"))?;
let addr = bound.local_addr();
let tls_on = bound.tls();
tracing::info!(
%addr,
anonymous_read = config.allow_anonymous,
cleartext_credentials = ?config.http_cleartext,
tls = tls_on,
"gunnar smart-http listener: info/refs, git-upload-pack, git-receive-pack, \
info/lfs/objects (batch, get, put, verify)"
);
// A32-14, stays a task: a listener; its handle is kept and it ends at the server's stop.
Some((tokio::spawn(bound.serve()), addr, tls_on))
}
};
// ── the scrape endpoint ────────────────────────────────────────────
//
// The SAME `Arc<ServerMetrics>` the heartbeat reads and both transports
// write. Two readings of one set of counters, never two sets: a scrape
// that disagreed with the heartbeat would leave an operator with no way
// to decide which one to believe during the one hour they need either.
//
// On the GIT runtime and not the control plane's: the whole point of
// this endpoint is that it still answers when the control plane is what
// is wedged, and putting it on the pool that starves would have made it
// useless in exactly that case.
let scrape = match config.metrics_bind {
None => None,
Some(bind) => {
let bound = crate::scrape::bind(bind, Arc::clone(&metrics))
.await
.with_context(|| format!("binding the metrics listener on {bind}"))?;
let addr = bound.local_addr();
// A32-14, stays a task: a listener; its handle is kept and it ends at the server's stop.
Some((tokio::spawn(bound.serve()), addr))
}
};
// ── the gRPC control plane, on a runtime of its own ────────────────
// ── replication slot: built EMPTY, and built EARLY ─────────────────
//
// Before the control plane, because the control plane takes a clone of
// it: `Mirror.SetReplication` has to work on a box that has never had a
// target and holds no repositories yet, which is every box on its first
// boot. And it captures THIS runtime — the git one — because a blocking
// send-pack belongs here and not on the control plane an operator is
// using while the box is in trouble.
#[cfg(feature = "replication")]
let replication = crate::mirror::ReplicationSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
tokio::runtime::Handle::current(),
);
// ── the outbound webhook slot: same shape, same reason ─────────────
//
// Built empty and early, for the same reason the replication slot is:
// the control plane takes a clone of it so `Webhooks.AddWebhook` works
// on a box that has never had an endpoint. Its workers do async POSTs,
// not blocking send-packs, but sharing the git runtime keeps them off
// the control plane an operator uses while the box is in trouble.
#[cfg(feature = "webhooks")]
let webhook = crate::webhook::WebhookSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
tokio::runtime::Handle::current(),
);
// The control plane's copy of the replication view, taken while the
// plane is built and filled below once the replicator exists.
let mut control_view: Option<crate::replication_view::ReplicationView> = None;
#[cfg(feature = "replication")]
let mut control_for_repair: Option<Arc<dyn gunnar_grpc::ControlPlane>> = None;
// The SAME grow machine `Admin.GrowStart` drives: a pulled `grow_go_ahead` (G16-d).
#[cfg(feature = "replication")]
let mut grow_for_repair: Option<Arc<crate::grow::GrowMachine>> = None;
let (grpc, grpc_shutdown, grpc_endpoint, control_runtime) = if config.no_grpc {
tracing::warn!(
"the gRPC control plane is disabled; repositories, keys and grants can only be \
administered from this host's command line"
);
(None, None, None, ControlRuntime::default())
} else {
let workers = config
.control_plane_workers
.unwrap_or_else(control_plane_workers);
// **The calls on one runtime, the connections on another**
// (R5.1-G). The handler runtime is held by a guard from the moment
// it exists, so the `?` below cannot drop a bare runtime here (see
// `ControlRuntime`).
let mut handler_runtime = ControlRuntime::new(
tokio::runtime::Builder::new_multi_thread()
.worker_threads(workers)
.thread_name("gunnar-control-call")
.enable_all()
.build()
.context("building the control plane's handler runtime")?,
);
let handlers = handler_runtime
.handle()
.context("the control plane's handler runtime was gone as soon as it was built")?;
let control_runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(CONTROL_CONNECTION_WORKERS)
.thread_name("gunnar-control")
.enable_all()
.build()
.context("building the control plane's runtime")?;
let grpc_bind = resolved_grpc_bind;
let grpc_bind_device = config.grpc_bind_device.clone();
let web_relying_parties = config
.web_relying_parties
.iter()
.map(|party| {
party
.parse::<gunnar_grpc::webauthn::RelyingParty>()
.map_err(|why| anyhow::anyhow!("--web-relying-party: {why}"))
})
.collect::<Result<Vec<_>>>()?;
let ssh_hint = ssh.local_addr().to_string();
// **ONE twin heartbeat registry, shared by the gRPC plane that
// RECORDS a twin's poll and the drain that READS what it applied.**
//
// It used to be created inside `ServiceContext` and owned only
// there, so the grow world could not see it — and the drain proved
// itself instead by pushing to the replication roster, which for a
// twin can never succeed (a twin serves no git). Built here so both
// halves hold the same `Arc`; two registries would be two answers
// to "what has the twin applied", and the one in front of the
// unmount would be the empty one.
let heartbeats = Arc::new(gunnar_grpc::heartbeat::Registry::default());
// ── the runtime data-set growth (`crate::grow`) ────────────────
//
// Over the ONE admission gate the fronts share (the drain pauses
// it) and these stores (the identity handle, the close/reopen).
// With `grow-runtime` the world is the box; without it the machine
// still answers `GrowStatus` honestly and refuses `GrowStart` by
// name.
#[cfg(feature = "grow-runtime")]
let grow_world: Arc<dyn crate::grow::World> = Arc::new(
crate::grow::world::BoxWorld::new(ssh.admission_gate(), Arc::clone(&stores))
// So the drain can prove the standbys are caught up once
// the gate is shut and nothing is in flight — the only
// moment at which "everything is drained to the twin" is
// a true sentence. Without this the world falls back to
// `NoStandby` and a growth would proceed having proved
// nothing about a twin that exists.
.with_replication(Arc::clone(&replication))
// The twin's OWN reported position — the only thing on
// this box that says whether it holds the set.
.with_heartbeats(Arc::clone(&heartbeats)),
);
#[cfg(not(feature = "grow-runtime"))]
let grow_world: Arc<dyn crate::grow::World> = Arc::new(crate::grow::world::NoEngine);
// **The go-ahead policy is the metered half's, and only the metered
// half's.** AUDITED 2026-09-18: this line read
// `GrowMachine::new(grow_world)` on every build, which installs
// `OperatorPolicy` — and that policy refuses every signer but
// `operator` by name. monetize signs its go-ahead `signer:
// "monetize"`, so its orchestrator reached `phase::STARTING`,
// called `Admin.GrowStart` and was refused AFTER it had reserved
// the spend: front-driven growth was dead at step 3 and had never
// worked. `MonetizePolicy` verifies that signature under the key
// the box already holds from `--monetize-pubkey` (the same one
// `EntitlementSlot::open` above was given), and it takes an
// `operator` go-ahead exactly as before. `cfg`, not a runtime
// branch: a community binary has no monetize verifier in it at all.
#[cfg(feature = "metered")]
let grow = match &config.metered {
Some(metered) => {
// Keyless when the key is fed and has not arrived: it is then
// applied here in place too (`feed::on_monetize_key`).
let policy = Arc::new(match &metered.monetize_pubkey {
Some(key) => crate::grow::MonetizePolicy::new(key)
.map_err(|e| anyhow::anyhow!("the growth go-ahead policy: {e}"))?,
None => crate::grow::MonetizePolicy::awaiting(),
});
let taker = Arc::clone(&policy);
crate::feed::on_monetize_key(
"growth go-ahead policy",
Box::new(move |k| taker.set_key(k)),
);
crate::grow::GrowMachine::with_policy(grow_world, policy)
}
// A metered binary started without `--monetize-pubkey` serves
// Free (the warning three hundred lines up); it has no key to
// verify a `monetize` go-ahead with either, so it keeps the
// operator door and refuses monetize by name rather than
// pretending to a check it cannot make.
None => crate::grow::GrowMachine::new(grow_world),
};
#[cfg(not(feature = "metered"))]
let grow = crate::grow::GrowMachine::new(grow_world);
#[cfg(feature = "replication")]
{
grow_for_repair = Some(Arc::clone(&grow));
}
let mut plane = CoreControlPlane::new(Arc::clone(&stores))
// `Bootstrap.Restore`'s keys from the medium; the fed ones are read per ask.
.with_restore_keys(restore_keys.clone())
// `Repos.Create` asks the SAME gate the front doors ask (one implementation):
// a disk that refuses a push refuses a new repository with the
// same line.
.with_headroom(Arc::clone(&disk_gate))
.with_grow(grow)
// The in-flight count `Mirror.FlushReplication` reports (T11).
.with_admission(ssh.admission_gate())
// The SAME bus the SSH front door publishes a push onto, so a
// `MirrorSynced` and a `RefUpdated` reach one `Events.Watch`
// stream. The plane publishes it from the mirror queue's worker
// when a sync completes — `Mirror.Sync` itself only enqueues,
// and an event announcing a fetch that has not happened would
// be worse than no event.
.with_bus(Arc::clone(&bus))
// `Bootstrap.Status` names the role before the join (a twin has no banner).
.with_twin_role(config.twin_role)
// **`serving_git` is a reading of THIS socket.** Both addresses,
// so the status door can say what was asked for and what was
// taken: on a twin those differ, because the branch above binds
// loopback:0 whatever `--ssh-bind` named, and a box that reports
// `git serving` while its :2222 sends an RST is a health line
// that cannot fail.
.with_git_listener(crate::readiness::GitListener {
asked: config.ssh_bind,
bound: ssh.local_addr(),
});
// The plane publishes into the SAME view the band reads, so
// `Admin.Stats` and the console can never disagree about whether a
// standby is taking bytes (one implementation).
// The WRITE side. `replication_view` just below is the READ side
// one slot, two halves, and neither is the other's twin —
// a `SetReplication` changes what is running, a `Stats` reads what
// it did.
#[cfg(feature = "replication")]
{
plane = plane.with_replication(Arc::clone(&replication));
}
#[cfg(feature = "webhooks")]
{
plane = plane.with_webhooks(Arc::clone(&webhook));
}
#[cfg(feature = "metered")]
if let Some(slot) = &entitlement {
plane = plane.with_entitlement(Arc::clone(slot));
}
let plane_view = plane.replication_view();
if let Some(bytes) = config.max_object_bytes {
plane = plane.with_max_object_bytes(bytes);
}
// **An import the last process did not live to finish is settled and
// resumed** (import-big, 2026-10-26), on the control runtime the
// mirror queue's lane lives on. A twin originates no sync.
if !config.twin_role {
let _on = control_runtime.enter();
let resumed = plane.resume_interrupted_mirrors();
if resumed > 0 {
tracing::info!(target: "gunnar::git", resumed, "mirror syncs interrupted by the last stop were resumed");
}
}
// The trust root, resolved BEFORE the listener is spawned, for the
// reason `--http-tls-*` is resolved before its bind: a control
// plane that was asked for client certificates and could not mint
// its own identity must be a startup error naming the cause, never
// a listener that came up accepting anyone.
let grpc_trust = match config.grpc_tls {
None => None,
Some(GrpcTls::SelfMinted) => {
let pki = stores
.pki()
.context("--grpc-tls-self needs the secret store; no PKI is available")?;
let issuer = pki
.issuer()
.context("reading the issuer for --grpc-tls-self")?
.context(
"--grpc-tls-self needs an issuer and none exists; the first-start \
mint above must have failed — its warning says why",
)?;
// The same SAN reasoning as the HTTP listener's: loopback
// always, plus a concrete bind address when there is one.
let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
let ip = grpc_bind.ip();
if !ip.is_unspecified() && !ip.is_loopback() {
hosts.push(ip.to_string());
}
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the gRPC server certificate")?;
let key_pem = issued
.key_pem
.clone()
.context("the issuer minted a server certificate but returned no key")?;
tracing::info!(
fingerprint = %issued.fingerprint,
sans = ?hosts,
"gRPC control plane: mutual TLS on. Clients must present a certificate \
this server's issuer signed AND registered (`gunnar control bootstrap`, \
`gunnar key add`)"
);
Some(gunnar_grpc::Trust {
// The issuer is both halves here: the CA every client
// certificate must chain to, and the second link of
// the chain this listener presents. Beside it, the
// CENTRAL monetize's attach CA when this estate
// attached from another cloud (f-grpc-tls): the
// central dials in with a client identity under it.
#[cfg(feature = "wg-client")]
client_ca: crate::fed_principals::grpc_client_cas(
&issuer.cert_pem,
stores.root(),
)
.into_bytes(),
// No fed attach CA without the installer's feed (wg-client).
#[cfg(not(feature = "wg-client"))]
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", issued.cert_pem, issuer.cert_pem).into_bytes(),
key: key_pem.into_bytes(),
})
}
};
// Taken from the trust root itself rather than from
// `config.grpc_tls`, so the field `Admin.Health` reports is the
// one the listener is actually built with.
let grpc_mutual_tls = grpc_trust.is_some();
// ── the MUTUAL REPLICATION LISTENER (lane T9) ──────────────────────
//
// The door a twin follows this box through. Resolved HERE, before
// anything is spawned, for the reason the main trust root is: a box
// that should serve its twin and cannot mint the identity to do it
// must fail to start NAMING the cause, not come up with a twin that
// retries into nothing for ever. `grpc_replication_bind` unset derives
// it from the main bind (off loopback ⇒ port + 1), which is what lets
// the one medium carry no flag.
let replication_root = stores.root().to_path_buf();
let replication = {
// The ONE derivation (`client_signer::replication_address`), so the
// operator side that dials it cannot drift from what is bound here.
let derived = (!grpc_bind.ip().is_loopback() && grpc_bind.port() != 0)
.then(|| {
gunnar_grpc::client_signer::replication_address(&grpc_bind.to_string()).ok()
})
.flatten()
.and_then(|at| at.parse::<SocketAddr>().ok());
// **Where it binds** (owner 2026-09-22, amended 2026-09-24): on the
// derived (eth1) address for the process lifetime, and ALSO on
// `wg0`'s address while the tunnel is up (`replication_door`); an
// explicit `--grpc-replication-bind` is never widened.
let explicit = config.grpc_replication_bind.is_some();
match config.grpc_replication_bind.or(derived) {
None => None,
Some(base) => {
let bind = base;
let pki = stores.pki().with_context(|| {
format!("the mutual replication listener on {bind} needs the secret store; no PKI is available")
})?;
let issuer = pki.issuer().context("reading the issuer for the replication listener")?.with_context(|| {
format!(
"the mutual replication listener on {bind} needs this box's issuer and there is \
none — the first-start mint above must have failed, and its warning says why. \
Without it no twin of this box can ever follow it."
)
})?;
let mut hosts = vec![
gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
"localhost".to_owned(),
"127.0.0.1".to_owned(),
];
for ip in [bind.ip(), base.ip()] {
if !ip.is_unspecified()
&& !ip.is_loopback()
&& !hosts.contains(&ip.to_string())
{
hosts.push(ip.to_string());
}
}
// Minted at EVERY start, so a restart is a renewal and a
// lapse needs a year of uninterrupted uptime.
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the replication listener's certificate")?;
let key_pem = issued.key_pem.clone().context(
"the issuer minted the replication certificate but returned no key",
)?;
Some((
base,
explicit,
bind,
gunnar_grpc::Trust {
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", issued.cert_pem, issuer.cert_pem)
.into_bytes(),
key: key_pem.into_bytes(),
},
))
}
}
};
let control = Arc::new(plane);
control_view = Some(plane_view);
#[cfg(feature = "replication")]
{
control_for_repair = Some(control.clone());
}
let bus_for_grpc = Arc::clone(&bus);
// ── ONE admission gate on ALL THREE front doors ────────────────
//
// `Repos.ArchiveOut` streams a whole repository's stored bytes down
// the control-plane channel — a transfer in everything but name — so
// it takes a permit from the SAME `Arc<Admission>` the SSH and HTTP
// git fronts hold (`ssh.admission_gate()`), not a second gate (one
// implementation). The *cheap* control RPCs get no gate on purpose: the control
// plane runs on its own runtime, so their walks starve
// nothing but themselves. Captured here and moved into the control
// runtime's spawn below.
let grpc_transfer_gate = crate::grpc_gate::control_transfer_gate(ssh.admission_gate());
let browse_flag = stores.browse();
// ── the durable audit trail (GUN-044) ──────────────────────────
//
// Opened HERE, on the startup path, because `AuditChain::open`
// verifies the existing chain and an operator has to be told about
// a broken one at the moment the server starts — not the first time
// somebody runs a verification months later.
//
// A chain that will not open does NOT stop the server. The control
// plane still authenticates, still authorises and still writes the
// `tracing` record it always did; what it loses is the durable
// copy, and it says so at `error!` with the reason. Refusing to
// serve would turn a full disk — or a chain somebody has already
// tampered with — into an outage, which hands whoever tampered
// with it exactly what they wanted.
let audit_ledger = {
// **Off the stripe.** The trail records who changed what on
// this box, which is exactly the question asked the morning
// after the four striped data disks have gone — so a copy that
// dies with them answers nobody. `ServerStores::audit_path`
// adopts an existing chain from the control directory once and
// falls back to it unchanged if the slow disk will not take it.
let path = stores.audit_path();
// ── and it must SURVIVE A RESTORE (lane T11, 2026-09-21) ───
//
// D6: `log/` was a per-box path, so the twin never held the
// trail and `twin replay` brought a box back with no audit
// history at all — silently, reporting success. A system whose
// audit trail is destroyed by its own recovery procedure
// cannot answer the only question that will be asked.
//
// So the primary's chain is opened FRAMED: every entry is an
// `Append` transaction, the twin holds it, and a restored box
// continues the dead primary's chain from its own tip.
//
// A twin — and a box mid-restore — originates nothing, so it
// gets an ordinary chain at its OWN per-box path
// (`audit_path` decides the name), and this branch does not
// even try to frame it.
let originates = !stores.is_twin() && !crate::twin::is_restoring(stores.root());
let opened = if originates {
match gunnar_core::AuditChain::open_framed(&path) {
Ok(chain) => Ok(chain),
// **Never silent.** An unframable chain is the D6 bug
// coming back, so it is said in the words an operator
// needs, and the server still writes a local trail.
Err(err @ gunnar_core::AuditError::Unframed { .. }) => {
tracing::error!(
target: gunnar_grpc::AUDIT_TARGET,
path = %path.display(),
error = %err,
"this box's audit trail is NOT replicated: it will be written \
here and it will NOT survive a restore from the twin. Every \
other class of state comes back; this one will not."
);
gunnar_core::AuditChain::open(&path)
}
Err(err) => Err(err),
}
} else {
gunnar_core::AuditChain::open(&path)
};
match opened {
Ok(chain) => Some(crate::grpc_gate::control_audit_ledger(Arc::new(chain))),
Err(err) => {
tracing::error!(
target: gunnar_grpc::AUDIT_TARGET,
path = %path.display(),
error = %err,
"the control-plane audit chain could not be opened; this server \
will log its decisions and will NOT write them to a trail. If the \
chain is reported BROKEN, the file has been edited or truncated \
since gunnar wrote it."
);
None
}
}
};
let (tx, rx) = oneshot::channel();
// The listener is bound INSIDE the control runtime. A tokio socket
// registers with the I/O driver of the runtime that created it, so
// one bound out here would keep the git runtime's driver in the
// control plane's critical path, which is the exact coupling this
// change exists to remove.
let (bound_tx, bound_rx) = oneshot::channel::<Result<(SocketAddr, bool)>>();
// The box's jera ledger (RA3-5): the job queue's jobs and the long duties are rows in
// `<data>/jobs`, read by `nornir jobs`; one that cannot open is said, and they run unrecorded.
if let Err(e) = gunnar_common::jobs::set_ledger(&stores.root().join("jobs")) {
tracing::error!("ALERT jobs.ledger {e}: the jobs run, unrecorded");
}
let sweep_stores = Arc::clone(&stores);
let activity_stores = Arc::clone(&stores);
let wg_only = config.wg_only;
// A32-14, stays a task: the control plane on its own runtime; its handle is kept and joined at the stop.
let join = control_runtime.spawn(async move {
// Bound WITH PATIENCE: a control plane on a tailnet or VPN
// address is asked for before that interface exists on a cold
// boot (builder, every boot until 2026-09-10: six `EADDRNOTAVAIL`
// exits and six blind systemd restarts before the seventh try
// found the address). The wait is bounded and only for that
// one errno; everything else still fails at once.
let listener = match bind_when_the_address_arrives(
|| bind_grpc_listener(grpc_bind, grpc_bind_device.as_deref()),
GRPC_BIND_PATIENCE,
GRPC_BIND_RETRY_EVERY,
)
.await
.0
.with_context(|| match &grpc_bind_device {
Some(dev) => format!("binding the gRPC listener on {grpc_bind} (device {dev})"),
None => format!("binding the gRPC listener on {grpc_bind}"),
}) {
Ok(listener) => listener,
Err(err) => {
// Report the bind failure to `start`, which is the only
// place that can turn it into a startup error. Swallowing
// it here would leave `gunnar serve` announcing a control
// plane that never bound.
let _ = bound_tx.send(Err(err));
return Ok(());
}
};
let addr = match listener
.local_addr()
.context("reading the gRPC local address")
{
Ok(addr) => addr,
Err(err) => {
let _ = bound_tx.send(Err(err));
return Ok(());
}
};
let server_config = ServerConfig::default()
.with_bind(addr)
.with_ssh_bind_hint(ssh_hint)
.with_web_relying_parties(web_relying_parties)
.with_web_user_verification(if config.web_require_user_verification {
gunnar_grpc::config::UserVerification::Required
} else {
gunnar_grpc::config::UserVerification::Preferred
})
// Reported from the SAME value that decides whether the
// listener is built with a trust root, three lines below
// (`router_with_trust(…, grpc_trust.as_ref())`). Derived
// from anything else it would be a second opinion about
// whether the door is locked, and `Admin.Health`'s job
// here is to be the answer rather than to remember one.
.with_mutual_tls(grpc_mutual_tls);
// Every start, on purpose: a hole nobody is reminded of is a hole
// nobody fixes.
gunnar_grpc::warn_authentication_disabled(&server_config);
// Reported out rather than re-derived by the caller. `serve`
// prints this on stdout, and a second reader of "is auth on"
// is a second thing that can be wrong: the line it replaced
// said `UNAUTHENTICATED in v1` unconditionally, and had
// gone stale against a control plane that authenticates.
let authenticated = server_config.authentication_enabled();
let ctx = ServiceContext::new(control, bus_for_grpc, server_config)
// The one "this estate sells room" flag the push gate reads too.
.with_sells_room(Arc::clone(&sells_room))
// Every call on the handler runtime, the connections on
// this one (R5.1-G): no slow call can hold the worker that
// answers a client's keepalive PING.
.with_handler_runtime(handlers)
// The transfer-shaped RPC's permit, from the git
// fronts' own gate.
.with_transfer_gate(grpc_transfer_gate)
// The browse replica: the browse flag the mutator refusal reads —
// the same `Arc` the replication slot flips.
.with_browse_flag(browse_flag)
// The SAME registry the drain reads (see where it is built).
.with_heartbeats(Arc::clone(&heartbeats));
// The hash-chained trail, when there is one; `tracing` alone
// when there is not. Either way the gate's decisions also
// stamp who last changed something (2026-10-21, the admin's
// user list): `grpc_gate::activity_ledger`.
let ctx = ctx.with_audit_ledger(crate::grpc_gate::activity_ledger(
audit_ledger,
Arc::clone(activity_stores.activity()),
));
// **The identity core** (IDENTITY-HOST, 2026-10-27): edda's one
// copy of sign-up, passkey login, invites, sessions and step-up,
// over Gunnar's host; its records persist beside the control state.
// Its records fail CLOSED (RA5-9): a document that does not read stops the
// start by name; empty tables over it would revive the deactivated.
let identity = match crate::identity::try_build(
activity_stores.root(),
Arc::clone(&ctx.control),
&ctx.config,
config.identity_root_site,
) {
Ok(identity) => identity,
Err(why) => {
let _ = bound_tx.send(Err(anyhow::anyhow!("{why}")));
return Ok(());
}
};
activity_stores.set_identity(&identity);
let ctx = ctx.with_identity(identity);
// The forge door's `gunnar-repo` verb answers from here on.
let _ = repo_door_ctx.set(ctx.clone());
// **Company SSO's network**: the one way gunnar reaches an
// account's identity provider. `GUNNAR_SSO_ALLOW_LOOPBACK_HTTP=1`
// is the operator's test door (an IdP on this box's loopback,
// over http) and nothing else.
let sso_loopback = std::env::var("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP").is_ok_and(|v| v.trim() == "1");
if sso_loopback {
tracing::warn!("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP=1: an identity provider on this box's loopback is reachable over http (a test door)");
}
ctx.sso.set_allow_loopback_http(sso_loopback);
// A32-18: the egress is `sso-egress`'s (every appliance row names it); a build
// without it sets none, so an identity provider's call is refused by name.
#[cfg(feature = "sso-egress")]
ctx.sso.set_egress(std::sync::Arc::new(crate::sso_egress::UreqEgress::with_nets(
sso_loopback,
crate::sso_egress::system_lookup(),
ctx.sso.private_nets_handle(),
)));
#[cfg(not(feature = "sso-egress"))]
tracing::info!("company SSO has no way out in this build (gunnar-server's `sso-egress` is off): an identity provider's call is refused by name");
// **The operator's switch** (A34-16, default = AIRGAP): the wired egress is
// handed out only while the feed says `sso_egress=on` (or, on a box fed
// nothing, `GUNNAR_SSO_EGRESS=on`); applied now and on every feed after.
{
let env = std::env::var("GUNNAR_SSO_EGRESS").ok();
let on = crate::feed::sso_egress_on(crate::feed::Fed::current().as_ref(), env.as_deref());
ctx.sso.set_egress_on(on);
tracing::info!(target: "gunnar::audit", event = "sso.egress", on, "company SSO's network is {}", if on { "ON: account admins' identity providers may be dialled" } else { "off (airgap): no identity provider is dialled" });
let sso = std::sync::Arc::clone(&ctx.sso);
// The resolver a domain's DNS proof is read from (A34-5): the feed's, or
// `GUNNAR_SSO_DNS_RESOLVER` where the feed names none.
let resolver_env = std::env::var("GUNNAR_SSO_DNS_RESOLVER").ok();
if let Some(resolver) = crate::feed::sso_dns_resolver(crate::feed::Fed::current().as_ref(), resolver_env.as_deref()) {
ctx.sso.set_dns_resolver(&resolver);
}
nordisk_pairing::feedbus::on_published("sso: the operator's egress switch", move |feed| {
let fed = crate::feed::Fed::from_feed(feed);
// The resolver, live.
sso.set_dns_resolver(crate::feed::sso_dns_resolver(Some(&fed), resolver_env.as_deref()).as_deref().unwrap_or(""));
let on = crate::feed::sso_egress_on(Some(&fed), env.as_deref());
if on != sso.egress_on() {
tracing::info!(target: "gunnar::audit", event = "sso.egress", on, "company SSO's network turned {}", if on { "ON" } else { "off" });
sso.set_egress_on(on);
}
});
}
// **The operator's internal networks a company IdP may live
// on** (owner 2026-10-06): the installer's feed, applied now and
// on every feed after (LIVE); a list gunnar refuses is said and
// applied as none — today's guard.
{
let sso = std::sync::Arc::clone(&ctx.sso);
nordisk_pairing::feedbus::on_published(
"sso: the operator's identity-provider networks",
move |feed| {
let nets = match crate::feed::sso_private_nets(Some(&crate::feed::Fed::from_feed(feed))) {
Ok(nets) => nets,
Err(why) => {
tracing::warn!(target: "gunnar::audit", event = "sso.private_nets_refused", reason = %why, "the fed identity-provider networks were refused; none are admitted");
gunnar_grpc::sso::PrivateNets::default()
}
};
if nets != sso.private_nets() {
tracing::info!(target: "gunnar::audit", event = "sso.private_nets", nets = %nets, "identity providers on the operator's internal networks: {}", if nets.is_empty() { "none (public addresses only)".to_owned() } else { nets.to_string() });
sso.set_private_nets(nets);
}
},
);
}
// **The transaction log's default retention** (h-txlog-2x): a PRIMARY
// retains its log behind its slowest REPORTING twin, minus a window, every
// ten minutes — never with no twin, never past a silent or unheard one, and
// never on a twin (its log is its only copy; the pass says so and does
// nothing). `GUNNAR_TXLOG_RETAIN=off` turns it off.
// A32-14, stays a task: a supervised async loop (a panicked pass is said) on the server's runtime; it ends with it.
tokio::spawn(gunnar_grpc::retention::run(ctx.clone()));
// **The account audit logs' retention** (2026-10-05): a year by
// default, swept every six hours on a primary.
// A jera job (A32-14) that ends with the stores; one that cannot start is said.
if let Err(e) = crate::account_audit_sweep::start(&sweep_stores, crate::account_audit_sweep::EVERY) {
tracing::error!(target: "gunnar::audit", "ALERT {} no sweep ({e}): the account audit logs are not pruned on this run", crate::account_audit_sweep::SWEEP_JOB);
}
// The replication listener: bound BEFORE the address is reported,
// so a port that will not bind is a startup error and not a twin
// that retries into a closed door. The same context — the same
// gate, grants and audit trail — on a door that demands a
// certificate at the handshake.
// One stop for the replication door, whichever address it is on
// (it may rebind when the tunnel comes up at runtime).
let (replication_stop_tx, replication_stop_rx) = tokio::sync::watch::channel(false);
match replication {
Some((base, explicit, bind, trust)) => {
let listener = match tokio::net::TcpListener::bind(bind).await {
Ok(listener) => listener,
Err(err) => {
let _ = bound_tx.send(Err(anyhow::Error::new(err).context(format!(
"binding the mutual replication listener on {bind} (the door this box's twin follows it through)"
))));
return Ok(());
}
};
let bound = listener.local_addr().unwrap_or(base);
let recorded = if wg_only { record_replication_door_wg_only(&replication_root, bound.port(), None) } else { record_replication_door(&replication_root, bound, None) };
if let Err(err) = recorded {
let _ = bound_tx.send(Err(err));
return Ok(());
}
// Built once up front so a broken trust is a STARTUP error;
// every rebind builds its own from the same trust.
if let Err(err) = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, Some(&trust)) {
let _ = bound_tx.send(Err(err.context("configuring the mutual replication listener")));
return Ok(());
}
tracing::info!(
%bind, suite = gunnar_grpc::preferred_bulk_suite(),
"mutual replication listener up: TLS 1.3, a client certificate from this box's issuer required"
);
// A32-14, stays a task: the replication listener's accept loop, ended by the server's stop.
tokio::spawn(replication_door(
listener,
base,
explicit,
wg_only,
ctx.clone(),
grpc_request_id,
trust,
replication_root.clone(),
replication_stop_rx,
));
}
None => {
if let Err(err) = crate::twin_tls::record_replication_port(&replication_root, None) {
let _ = bound_tx.send(Err(err));
return Ok(());
}
}
}
if bound_tx.send(Ok((addr, authenticated))).is_err() {
// `start` gave up on us; do not leave a listener running.
return Ok(());
}
if wg_only {
// **--wg-only** (owner 2026-10-04): the bound — utility —
// listener serves nothing; the control plane answers on the
// tunnel's address only, following it (`control_doors`).
gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref())
.context("configuring mutual TLS on the gRPC control plane")?;
let (control_stop_tx, control_stop_rx) = tokio::sync::watch::channel(false);
let serve_control = move |l: tokio::net::TcpListener, mut stop: tokio::sync::watch::Receiver<bool>| {
let router = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref());
async move {
let router = match router {
Ok(r) => r,
Err(err) => {
tracing::error!(%err, "the control plane's tunnel door could not be configured");
return;
}
};
let at = l.local_addr().map(|a| a.to_string()).unwrap_or_default();
if let Err(err) = router
.serve_with_incoming_shutdown(grpc_incoming(l), async move {
let _ = stop.wait_for(|s| *s).await;
})
.await
{
tracing::error!(%err, %at, "a control plane door stopped");
}
}
};
// A32-14, stays a task: the control doors' accept loop; its handle is kept and it stops at the stop signal.
let doors = tokio::spawn(control_doors(listener, addr, true, crate::tunnel_addr::subscribe(), control_stop_rx, serve_control));
let _ = rx.await;
// One stop for every door.
let _ = replication_stop_tx.send(true);
let _ = control_stop_tx.send(true);
let _ = doors.await;
Ok(())
} else {
gunnar_grpc::router_with_trust(ctx, grpc_request_id, grpc_trust.as_ref())
.context("configuring mutual TLS on the gRPC control plane")?
.serve_with_incoming_shutdown(
grpc_incoming(listener),
async {
let _ = rx.await;
// One stop for both doors.
let _ = replication_stop_tx.send(true);
},
)
.await
.context("the gRPC control plane stopped")
}
});
let (addr, grpc_authenticated) = match bound_rx.await {
Ok(Ok(pair)) => pair,
Ok(Err(err)) => {
control_runtime.shutdown_background();
return Err(err);
}
Err(_) => {
control_runtime.shutdown_background();
bail!(
"the gRPC control plane task ended before it reported an address; the \
listener on {grpc_bind} never came up"
);
}
};
tracing::info!(
%addr,
control_plane_workers = workers,
control_connection_workers = CONTROL_CONNECTION_WORKERS,
"gunnar grpc control plane listening on its own runtime (a control-plane \
stall cannot reach the git listener, and its calls run on a handler runtime \
that cannot hold the connections)"
);
let mut held = ControlRuntime::new(control_runtime);
held.hold(&mut handler_runtime);
(Some(join), Some(tx), Some((addr, grpc_authenticated)), held)
};
// ── the log level, changeable from here on ─────────────────────────
//
// Started here rather than in `main` because it needs a runtime, and
// deliberately NOT held on `Server`: the task owns two signal streams
// and nothing else, and a level-control task that stopped with the
// server would be one more thing to get wrong at shutdown.
//
// It also has to exist for a NEGATIVE reason. The default disposition
// of both `SIGUSR1` and `SIGUSR2` is to TERMINATE the process, so
// before this an operator who reached for the usual signal killed the
// server they were debugging.
gunnar_common::logging::spawn_signal_control();
// ── the counter heartbeat ──────────────────────────────────────────
let heartbeat = config.metrics_interval.map(|interval| {
let metrics = Arc::clone(&metrics);
// A32-14, stays a task: its handle is kept and aborted at the server's stop.
tokio::spawn(async move { heartbeat(metrics, interval).await })
});
// ── replication, a SUBSCRIBER and nothing else ─────────────────────
//
// Started last, and started on the git runtime deliberately: the work
// it does is a git push over SSH, and the blocking part of it goes to
// the blocking pool. Nothing above this line knows it exists, which is
// the property that makes "replication can never fail a push" true by
// construction rather than by care.
#[cfg(feature = "replication")]
{
use crate::mirror::ReplicationConfig;
let replication_config = match config.replication.clone() {
Some(explicit) => explicit,
None => ReplicationConfig::load(stores.root())
.context("reading the replication configuration")?,
};
// An empty target list leaves the slot empty, which is still replication's
// "a server that replicates nowhere pays nothing": no replicator, no
// subscriber, no task. What it no longer means is that the operator
// has no way to add one.
replication
.apply(replication_config)
.context("starting replication")?;
}
// ── I11's passive heartbeat, if this box was asked for one ─────────
//
// Started AFTER the slot has its configuration, because the task's
// first act is to read the role out of it: a watch spawned before
// `apply` would read the pre-boot file and could leave immediately on a
// box that is about to become a standby.
//
// On the GIT runtime, with the replicator, and for the same reason: the
// probe is a blocking-shaped network wait, and the control plane is the
// surface an operator uses while a box is in trouble.
#[cfg(feature = "replication")]
let standby_watch = config.standby_watch.map(|watch| {
tracing::info!(
target: "gunnar::replication",
primary = %watch.primary,
probe_interval_ms = watch.probe_interval.as_millis() as u64,
promote_after = watch.promote_after,
earliest_takeover_ms = watch.earliest_takeover().as_millis() as u64,
"watching the primary; this box takes over by itself if it stops answering"
);
let slot = Arc::clone(&replication);
// A32-14, stays a task: its handle is kept and aborted at the server's stop.
tokio::spawn(async move { crate::standby_watch::watch(slot, watch).await })
});
// ── 5.4: the boot check, then the pull channel ─────────────────────
//
// After the control plane listens: the boot check reads this box's own
// `Bootstrap.Status`, and a recharge into this box goes through it.
#[cfg(feature = "replication")]
let repair = match (config.repair, grpc_endpoint) {
(Some(cfg), Some((bound, _))) => {
let own = crate::repair::own_addr(bound);
let mut entitlements = crate::repair::Entitlements::of(Arc::clone(&stores));
if let Some(control) = control_for_repair.clone() {
entitlements = entitlements.with_control(control);
}
#[cfg(feature = "metered")]
let mut entitlements = entitlements.with_slot(entitlement.clone());
if let Some(grow) = grow_for_repair.clone() {
entitlements =
entitlements.with_grow(crate::paid_growth::PaidGrowth::new(grow));
}
// The environment's relay; else (a SOLD estate's main, P1) the one the shop FED it,
// read at each mail — not ready until the feed names it, so a resend is refused
// by name before anything is minted.
#[cfg(feature = "fleet-pull")]
let postman: Option<Arc<dyn crate::postman::Postman>> =
match crate::postman::Smtp::from_env() {
Ok(Some(smtp)) => Some(Arc::new(smtp)),
// MAIL-RELAY: the relay the ONE box fed this main again goes over the lay's.
Ok(None) => {
let root = stores.root().to_path_buf();
Some(Arc::new(crate::postman::FedRelay {
fed: move || {
crate::mail_relay::over(crate::feed::Fed::current(), &root)
},
}))
}
Err(why) => {
tracing::warn!(target: crate::repair::TARGET, "no mail relay: a sign_in_again order is refused by name ({why})");
None
}
};
let entitlements = entitlements.with_twin(cfg.twin.clone());
#[cfg(feature = "fleet-pull")]
let entitlements = entitlements.with_postman(postman);
let stores = Arc::clone(&stores);
crate::repair::arm(cfg.clone(), own.clone(), tokio::runtime::Handle::current());
// A32-14, stays a task: the repair boot and the fleet pump, async on the server's runtime; its handle is kept and aborted at the stop.
Some(tokio::spawn(async move {
crate::repair::boot(Arc::clone(&stores), cfg.clone(), own.clone()).await;
// AIRGAP B5 + M12: only a `fleet-pull` build carries the pull channel (and
// the shared pump it runs on); any other build refuses a named one by name
// (`repair::resolve_fleet`) and pulls nothing.
#[cfg(not(feature = "fleet-pull"))]
let _ = (entitlements, own);
#[cfg(feature = "fleet-pull")]
if let Some(fleet) =
crate::repair::fleet_when_fed(cfg.fleet.clone(), crate::repair::FED_EVERY)
.await
{
let estate = crate::repair::ServerEstate::new(stores.root(), &cfg, own.clone())
.map_err(|why| tracing::warn!(target: crate::repair::TARGET, "the pull channel runs without a repair key: every order is refused by name ({why})"))
.ok();
let (_keep, stop) = tokio::sync::watch::channel(false);
crate::repair::pull_loop(stores, fleet, estate, entitlements, own, stop)
.await;
}
}))
}
(Some(_), None) => {
tracing::warn!(target: "gunnar::repair", "the boot check and the pull channel need the control plane, which is off");
None
}
(None, _) => None,
};
// Outbound webhooks: load webhooks.json and start delivering. An empty file
// (or none) leaves the slot empty — no subscriber, no task — the "a
// server with no webhooks pays nothing" property; a malformed file is a
// hard error here, so a typo is caught at boot and not per delivery.
#[cfg(feature = "webhooks")]
{
let webhook_config = crate::webhook::WebhookConfig::load(stores.root())
.context("reading the webhook configuration")?;
webhook
.apply(webhook_config)
.context("starting outbound webhook delivery")?;
}
// ── publish the counters, so something finally READS them ──────────
//
// One closure, two readers: `config.replication_view` is the console
// band's and `control_view` is `Admin.Stats`'s. Before this the
// counters existed and nothing looked at them, which is how every
// configured standby receiving NOTHING went unnoticed for weeks.
//
// The closure reads the slot on EVERY call rather than closing over one
// `Arc<Replicator>`. That is what keeps the band and `Admin.Stats`
// honest after a live `SetReplication`: a captured replicator would go
// on reporting the standby that was just removed, and report nothing at
// all about the one that was just added. With no target configured the
// slot is empty and the answer is `ReplicationReport::default()` —
// `configured: false`, which is a fact rather than an absence.
#[cfg(feature = "replication")]
for view in [Some(config.replication_view.clone()), control_view.clone()]
.into_iter()
.flatten()
{
let slot = Arc::clone(&replication);
view.fill(move || {
slot.replicator()
.as_deref()
.map(|replicator| {
// The cheap report, plus the ref number PEEKED from the
// last on-demand walk — never a walk of its own. This
// closure runs on every `Admin.Stats` and every band
// frame, so a reader that walked would be the cold-status
// "re-open every store every 60 s from a polling
// caller". `ref_coverage_cached` opens nothing; the
// number is whatever the most recent `--refs` warmed,
// and zeros until the first one.
let mut report = crate::replication_view::report(replicator);
report.refs = replicator.ref_coverage_cached();
report
})
.unwrap_or_default()
});
}
let _ = &control_view;
Ok(Server {
ssh: Some(ssh),
grpc,
grpc_shutdown,
grpc_endpoint,
control_runtime,
stores,
bootstrap_window_open,
host_key_fingerprint,
metrics,
heartbeat,
#[cfg(feature = "replication")]
standby_watch,
#[cfg(feature = "replication")]
repair,
bus,
#[cfg(feature = "http")]
http,
scrape,
#[cfg(feature = "replication")]
replication,
#[cfg(feature = "webhooks")]
webhook,
})
}
/// The event bus.
///
/// The same object `Events.Watch` streams from, exposed here so an
/// in-process caller (the replicator, and the tests that assert what it
/// publishes) can subscribe without a gRPC listener and a credential in
/// between.
pub fn bus(&self) -> &Arc<EventBus> {
&self.bus
}
/// The replicator, when this server has any standby configured.
///
/// Exposed so an operator surface, and `tests/replication.rs`, can read the
/// counters and the per-standby status without reaching into the tasks.
#[cfg(feature = "replication")]
pub fn replication(&self) -> Option<Arc<crate::mirror::Replicator>> {
self.replication.replicator()
}
/// The live slot, for a caller that RECONFIGURES rather than only observes:
/// `Mirror.SetReplication`'s in-process half and `tests/replication.rs`.
#[cfg(feature = "replication")]
pub fn replication_slot(&self) -> &Arc<crate::mirror::ReplicationSlot> {
&self.replication
}
/// The live webhook slot, for a caller that reconfigures outbound delivery:
/// `Webhooks.AddWebhook`'s in-process half and `tests/webhooks.rs`.
#[cfg(feature = "webhooks")]
pub fn webhook_slot(&self) -> &Arc<crate::webhook::WebhookSlot> {
&self.webhook
}
/// **The SSH host key this server serves**, `SHA256:…` — the fingerprint a
/// client pins, read off the key that is actually loaded rather than off a
/// file somebody remembers.
pub fn host_key_fingerprint(&self) -> &str {
&self.host_key_fingerprint
}
/// The address the SSH listener actually bound.
pub fn ssh_addr(&self) -> SocketAddr {
self.ssh().local_addr()
}
/// The address the scrape endpoint bound, if it was started.
///
/// `None` is the ordinary answer: nothing binds unless `--metrics-bind`
/// asked for it. A test that asked for port 0 learns its port here.
pub fn metrics_addr(&self) -> Option<SocketAddr> {
self.scrape.as_ref().map(|(_, addr)| *addr)
}
/// The address the smart-HTTP listener bound, if it was started.
#[cfg(feature = "http")]
pub fn http_addr(&self) -> Option<SocketAddr> {
self.http.as_ref().map(|(_, addr, _)| *addr)
}
/// Whether the smart-HTTP listener terminates TLS itself.
#[cfg(feature = "http")]
pub fn http_tls(&self) -> bool {
self.http.as_ref().is_some_and(|(_, _, tls)| *tls)
}
/// The SSH listener. Present for the whole life of a `Server`; the `Option`
/// exists only so [`Server::shutdown`] can move it out past the [`Drop`]
/// impl.
fn ssh(&self) -> &BoundServer {
self.ssh
.as_ref()
.expect("the SSH listener is taken only by shutdown, which consumes the Server")
}
/// The address the gRPC listener actually bound, if it was started.
pub fn grpc_addr(&self) -> Option<SocketAddr> {
self.grpc_endpoint.map(|(addr, _)| addr)
}
/// The bound gRPC address together with whether that listener
/// authenticates, which is what `serve` prints.
///
/// One accessor rather than two, because the banner used to carry a
/// hardcoded `UNAUTHENTICATED in v1` beside a control plane that
/// authenticates. A caller that cannot get the address without the verdict
/// cannot print a stale one.
pub fn grpc_endpoint(&self) -> Option<(SocketAddr, bool)> {
self.grpc_endpoint
}
/// Was the unauthenticated enrolment window open when this server started?
///
/// What `serve` prints beside the gRPC address. `false` is the ordinary
/// answer, which is why the banner says nothing at all in that case.
pub fn bootstrap_window_open(&self) -> bool {
self.bootstrap_window_open
}
/// The stores behind both listeners.
pub fn stores(&self) -> &Arc<ServerStores> {
&self.stores
}
/// Serve until a stop is requested: a signal, or on the appliance the ACPI
/// power button ([`appliance_pid1::runtime::stop_request`]).
///
/// **Not `ctrl_c()`.** That is `SIGINT` alone, and nothing in an appliance's
/// life sends `SIGINT`: `podman stop` sends `SIGTERM` and every service
/// manager sends `SIGTERM`. A cloud's soft stop (UpCloud `stop`, QEMU
/// `system_powerdown`) is not a signal at all: it is an ACPI power-button
/// press the kernel reports on the `acpi_event` generic-netlink group
/// (corrected 2026-09-28; this said it became `SIGTERM`, and a soft stop
/// then waited out the cloud's timeout and was hard-stopped). PID 1 also
/// gets **no default disposition** for a signal it has not handled — the
/// kernel discards it. See edda's `appliance_pid1::net`.
pub async fn wait(self) -> Result<()> {
let signal = appliance_pid1::runtime::stop_request().await;
tracing::info!(signal, "shutting both listeners down");
self.shutdown().await
}
/// Ask both listeners to stop, and wait for them.
///
/// # Why the gRPC wait is bounded (measured 2026-08-03)
///
/// `serve_with_incoming_shutdown` is *graceful*: it stops accepting and
/// then waits for every in-flight request to finish. `Events.Watch` (the event stream)
/// is server-streaming and by design **never finishes**; it ends when the
/// subscriber goes away. So one connected UI made this wait forever, and
/// `gunnar serve` could not be stopped with Ctrl-C at all. Measured: a test
/// binary wedged for eleven minutes with the watch connection still
/// `ESTAB`, no CPU, no progress.
///
/// Graceful is still tried first — an in-flight `Repos.Create` should be
/// allowed to land — but after [`GRPC_SHUTDOWN_GRACE`] the task is aborted
/// and the process is free to exit. A watcher losing its stream at shutdown
/// is what a watcher is built to handle. A server that cannot be stopped
/// is not.
pub async fn shutdown(mut self) -> Result<()> {
// One last line before the listeners go, so the final numbers are in
// the log rather than only in a process that is about to exit.
if let Some(beat) = self.heartbeat.take() {
beat.abort();
}
// I11's watchdog goes with the listeners. A watch that outlived the
// server it belongs to could promote a box that is shutting down —
// writing `role: primary` to disk as its last act, so the NEXT boot
// comes up writable. Aborting it here is what keeps a shutdown a
// shutdown.
#[cfg(feature = "replication")]
if let Some(watch) = self.standby_watch.take() {
watch.abort();
}
#[cfg(feature = "replication")]
if let Some(repair) = self.repair.take() {
repair.abort();
}
// Replication stops with the server. An in-flight attempt is abandoned
// rather than waited for: the job it was draining is still in the
// queue's shape ("converge this repository") and the next start
// re-derives it from the repository itself.
#[cfg(feature = "replication")]
self.replication.stop();
#[cfg(feature = "webhooks")]
self.webhook.stop();
self.log_metrics();
if let Some(tx) = self.grpc_shutdown.take() {
let _ = tx.send(());
}
if let Some(join) = self.grpc.take() {
let abort = join.abort_handle();
match tokio::time::timeout(GRPC_SHUTDOWN_GRACE, join).await {
Ok(Ok(result)) => result?,
Ok(Err(err)) if err.is_cancelled() => {}
Ok(Err(err)) => return Err(anyhow::anyhow!("the gRPC task panicked: {err}")),
Err(_elapsed) => {
tracing::warn!(
grace = ?GRPC_SHUTDOWN_GRACE,
"the gRPC control plane did not stop gracefully; almost certainly an \
open Events.Watch stream, which by design never ends. Aborting it so \
the process can exit."
);
abort.abort();
}
}
}
// `shutdown_background`, not a drop: `shutdown` is an `async fn` and may
// well be running on a runtime thread, where dropping a `Runtime`
// panics. The listener is already stopped or aborted above, so there is
// nothing left worth blocking for. The `ControlRuntime` guard's own
// `Drop` would do this too; doing it explicitly here just tears the
// control plane down before the SSH listener rather than after.
self.control_runtime.shutdown();
if let Some(ssh) = self.ssh.take() {
ssh.shutdown("gunnar is shutting down")
.await
.context("stopping the SSH listener")?;
}
Ok(())
}
/// What the SSH admission gate has done since start-up.
///
/// Here so an operator surface (and `tests/`) can read the saturation
/// counters without reaching into `gunnar-ssh`.
pub fn admission(&self) -> gunnar_ssh::AdmissionStats {
self.ssh().admission()
}
/// Every counter this server keeps, read now.
///
/// One reading of one set: the SSH listener and the git transport write to
/// the same block, so `connections_refused` and `pushes_rejected` are facts
/// about the same server rather than two crates' private tallies.
pub fn metrics(&self) -> MetricsSnapshot {
self.metrics.snapshot()
}
/// The counter block itself, for a caller that wants to keep a handle to it.
pub fn metrics_handle(&self) -> &Arc<ServerMetrics> {
&self.metrics
}
/// Write the counter line now, without waiting for the next heartbeat.
pub fn log_metrics(&self) {
emit_metrics(&self.metrics.snapshot());
}
}
impl Drop for Server {
fn drop(&mut self) {
if let Some(beat) = self.heartbeat.take() {
beat.abort();
}
// Same reason as in `shutdown`: a watchdog that survives its server
// could promote a box on its way out.
#[cfg(feature = "replication")]
if let Some(watch) = self.standby_watch.take() {
watch.abort();
}
#[cfg(feature = "replication")]
if let Some(repair) = self.repair.take() {
repair.abort();
}
#[cfg(feature = "replication")]
self.replication.stop();
#[cfg(feature = "webhooks")]
self.webhook.stop();
// A `Server` dropped without `shutdown` (a test that panicked, say)
// would otherwise take its control runtime down by a blocking `Drop`,
// which panics on a runtime thread and would bury the real failure. The
// `ControlRuntime` field's own `Drop` already routes through
// `shutdown_background`; this is that same call, made explicit.
self.control_runtime.shutdown();
}
}
/// **Say, at every start, whether the unauthenticated enrolment door is open.**
///
/// Silence here is the disease this repository keeps finding. A server standing
/// with an open door is a server anybody who can reach the control port can
/// take; an operator must never have to guess which state their box is in, and
/// on an unenrolled server there is by definition no credential with which to
/// ask `Admin.Health`.
///
/// Split out of [`Server::start`] and given both readings rather than taking
/// them, so every branch can be driven from a test on any box. A warning that
/// reads its own host can only ever be seen in the state that host happens to
/// be in.
///
/// Returns the line it emitted, so a test asserts on the **text an operator
/// sees** rather than on a boolean that proves nothing about the log.
pub fn announce_readiness(
readiness: &Readiness,
data_root: &std::path::Path,
grpc_bind: SocketAddr,
no_grpc: bool,
) -> String {
match readiness {
Readiness::Ready => {
let line = format!("identity table ready at {}", data_root.display());
tracing::info!(data_root = %data_root.display(), "{line}");
line
}
Readiness::AwaitingBootstrap if no_grpc => {
let line = format!(
"NO IDENTITY HAS EVER BEEN ENROLLED at {}, and the gRPC control plane is \
DISABLED, so there is no way to enrol one. Nobody can reach this server. Start \
without --no-grpc and call Bootstrap.Enroll, or pass --admin-key.",
data_root.display()
);
tracing::error!(data_root = %data_root.display(), "{line}");
line
}
Readiness::AwaitingBootstrap => {
let line = format!(
"NO IDENTITY HAS EVER BEEN ENROLLED at {}. Bootstrap.Enroll on {grpc_bind} is \
open to ANY caller that can reach it, with NO credential, and it creates a \
gunnar-admin. It closes permanently on the first success and deleting \
identities does not reopen it. Enrol now, or bind the control plane where \
nobody else can reach it.",
data_root.display()
);
tracing::warn!(
data_root = %data_root.display(),
%grpc_bind,
bootstrap_window = "OPEN",
"{line}"
);
line
}
Readiness::SealedAndEmpty { record } => {
let who = match record {
Some(record) => format!(
"was bootstrapped by \"{}\" at {}",
record.principal_id, record.at
),
None => "holds principals but no keys and carries no bootstrap record, so it \
was written before that record existed, and"
.to_owned(),
};
let line = format!(
"this server {who} its identity table at {} now holds no keys. NOBODY CAN \
AUTHENTICATE against it, and the enrolment window stays shut — deleting \
identities does not reopen it. Add a key with `gunnar key add` while the \
server is down.",
data_root.display()
);
tracing::error!(
data_root = %data_root.display(),
bootstrap_window = "SHUT",
"{line}"
);
line
}
}
}
/// **Turn `--grpc-bind` into the address the listener binds.**
///
/// The one caller of [`BindSpec::resolve_with`] in the server, and the one
/// place `appliance_pid1::net::interface_ipv4` is wired to it. An address resolves to itself
/// and never touches netlink; an interface is asked, once, and a refusal stops
/// the start rather than widening the bind — the whole point of the spelling.
/// Returns the address to bind, or `None` when the named interface exists and
/// holds no IPv4 — see [`BindSpec::resolve_with`] for why that is `Ok` and not
/// an error, and what the caller owes it.
/// **Bind the control plane's listener — on a DEVICE when the spelling named
/// one.** `eth1:50051` resolves to the address eth1 holds, and an address alone
/// is not the interface: MEASURED 2026-09-05 on the KVM bench rig (two slirp
/// NICs, both 10.0.2.15), the listener bound to eth1's address answered on
/// eth0's forward and refused eth1's, because the kernel routes by address and
/// both interfaces held it. `SO_BINDTODEVICE` makes the socket the interface's
/// — the same call the DHCP client makes for its broadcast socket — so the
/// declaration `--grpc-bind eth1:50051` means eth1 on every machine, not just on
/// one whose subnets happen to differ. An address spelling (`None`) binds as
/// before. The socket is made non-blocking and handed to the control runtime's
/// I/O driver by `from_std`, on the runtime that calls this.
fn bind_grpc_listener(addr: SocketAddr, device: Option<&str>) -> Result<tokio::net::TcpListener> {
use socket2::{Domain, Protocol, Socket, Type};
let domain = if addr.is_ipv6() {
Domain::IPV6
} else {
Domain::IPV4
};
let sock = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
.context("opening the gRPC listener socket")?;
sock.set_reuse_address(true)
.context("SO_REUSEADDR on the gRPC listener")?;
if let Some(dev) = device {
sock.bind_device(Some(dev.as_bytes()))
.with_context(|| format!("SO_BINDTODEVICE {dev} on the gRPC listener"))?;
}
sock.bind(&addr.into())
.with_context(|| format!("bind {addr}"))?;
sock.listen(1024).context("listen on the gRPC listener")?;
sock.set_nonblocking(true)
.context("non-blocking gRPC listener")?;
tokio::net::TcpListener::from_std(sock.into())
.context("registering the gRPC listener with the runtime")
}
/// How long `serve` waits for the gRPC bind address to turn up on an interface
/// before it gives up. Thirty seconds covers a tailnet or WireGuard interface
/// that comes up after this service on a cold boot; anything slower is a box
/// with a real problem, and the bounded exit hands it to the supervisor.
pub const GRPC_BIND_PATIENCE: std::time::Duration = std::time::Duration::from_secs(30);
/// The pause between two bind attempts while waiting for the address.
pub const GRPC_BIND_RETRY_EVERY: std::time::Duration = std::time::Duration::from_millis(500);
/// `EADDRNOTAVAIL` and nothing else. That errno means "no interface on this
/// box has that address", which on a cold boot means *yet*. `EADDRINUSE`, a
/// bad device, a permission refusal: none of those get better by waiting.
fn bind_failed_because_the_address_is_not_here_yet(err: &anyhow::Error) -> bool {
err.chain()
.filter_map(|cause| cause.downcast_ref::<std::io::Error>())
.any(|io| io.kind() == std::io::ErrorKind::AddrNotAvailable)
}
/// Run `attempt` until it binds, retrying only while the failure is the
/// address-not-here-yet errno, every `every`, for at most `patience`.
///
/// Returns the final result AND the number of retries taken, so a caller (and
/// the test) can tell a first-try bind from a waited-for one.
async fn bind_when_the_address_arrives<T, F>(
mut attempt: F,
patience: std::time::Duration,
every: std::time::Duration,
) -> (Result<T>, u32)
where
F: FnMut() -> Result<T>,
{
let started = std::time::Instant::now();
let mut retries = 0u32;
loop {
match attempt() {
Ok(bound) => {
if retries > 0 {
tracing::info!(
retries,
waited_ms = started.elapsed().as_millis() as u64,
"the gRPC bind address arrived; the control plane is bound"
);
}
return (Ok(bound), retries);
}
Err(err) if bind_failed_because_the_address_is_not_here_yet(&err) => {
if started.elapsed() >= patience {
return (
Err(err.context(format!(
"the address was on no interface for {} s ({retries} retries)",
patience.as_secs()
))),
retries,
);
}
if retries == 0 {
tracing::warn!(
patience_s = patience.as_secs(),
"the gRPC bind address is on no interface yet (EADDRNOTAVAIL); waiting for \
it. A tailnet or VPN interface that comes up after this service is the \
usual cause"
);
}
retries += 1;
tokio::time::sleep(every).await;
}
Err(err) => return (Err(err), retries),
}
}
}
#[cfg(test)]
mod grpc_bind_patience_tests {
use super::*;
use std::time::Duration;
fn not_here_yet() -> anyhow::Error {
anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrNotAvailable))
.context("bind 100.64.0.1:50061")
}
/// The address arrives on the third try: bound, and the retry count says
/// the wait was real (a wrapper that never retried would return 0 here).
#[test]
fn an_address_that_arrives_later_is_bound_and_the_retries_are_counted() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut left = 2;
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| {
if left > 0 {
left -= 1;
Err(not_here_yet())
} else {
Ok(7u8)
}
},
Duration::from_secs(5),
Duration::from_millis(1),
));
assert_eq!(bound.unwrap(), 7);
assert_eq!(retries, 2);
}
/// Any other errno fails on the first try, with zero retries: waiting on
/// `EADDRINUSE` would hide a second server on the port for thirty seconds.
#[test]
fn an_address_in_use_is_not_waited_for() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut calls = 0;
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| -> Result<u8> {
calls += 1;
Err(anyhow::Error::from(std::io::Error::from(
std::io::ErrorKind::AddrInUse,
)))
},
Duration::from_secs(5),
Duration::from_millis(1),
));
assert!(bound.is_err());
assert_eq!(retries, 0);
assert_eq!(calls, 1);
}
/// The patience is a bound: an address that never comes gives up with the
/// wait named in the error, not a hang.
///
/// On a busy box (GUN-FIELD-TEST, oden at load 130: `retries=1`) the thread is
/// descheduled for longer than the patience; the first attempt below is that box,
/// made on purpose, so the test says the same thing on an idle box and a loaded one.
#[test]
fn an_address_that_never_comes_gives_up_after_the_patience() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut calls = 0u32;
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| -> Result<u8> {
calls += 1;
if calls == 1 {
std::thread::sleep(Duration::from_millis(40));
}
Err(not_here_yet())
},
Duration::from_millis(30),
Duration::from_millis(5),
));
let msg = format!("{:#}", bound.unwrap_err());
assert!(msg.contains("on no interface for 0 s"), "{msg}");
// What holds on any box: every retry was one more attempt, and the first attempt
// that found the patience spent was the last one (no attempt after the bound).
// How MANY retries fit in 30 ms is the scheduler's, not the wait's.
assert_eq!(retries + 1, calls, "retries={retries} calls={calls}");
assert_eq!(
calls, 1,
"the patience was spent at the first attempt: no second one"
);
}
/// The real socket path: TEST-NET-1 is on no interface anywhere, and the
/// kernel's answer is the errno the wait is keyed on. This is the red that
/// proves the classifier reads the errno that `bind_grpc_listener` actually
/// produces, not one the test invented.
#[test]
fn the_real_bind_error_for_an_absent_address_is_classified_as_not_here_yet() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let err = bind_grpc_listener("192.0.2.1:0".parse().unwrap(), None)
.expect_err("TEST-NET-1 is on no interface");
assert!(
bind_failed_because_the_address_is_not_here_yet(&err),
"{err:#}"
);
let in_use = anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrInUse));
assert!(!bind_failed_because_the_address_is_not_here_yet(&in_use));
}
}
#[cfg(test)]
mod grpc_bind_device_tests {
use super::*;
/// A device that does not exist is refused BY NAME before anything binds —
/// the red that proves the device half is wired at all.
#[test]
fn a_device_the_machine_does_not_have_is_refused_by_name() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let err = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("nosuchdev0"))
.expect_err("a nonexistent device cannot be bound");
let msg = format!("{err:#}");
assert!(msg.contains("SO_BINDTODEVICE nosuchdev0"), "{msg}");
}
/// The address-only spelling binds exactly as before, and on the loopback
/// device by name too (the one device every box has).
#[test]
fn an_address_binds_and_so_does_the_loopback_device() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let plain = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), None).unwrap();
assert!(plain.local_addr().unwrap().port() != 0);
match bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("lo")) {
Ok(l) => assert!(l.local_addr().unwrap().port() != 0),
// Kernels before 5.7 gate SO_BINDTODEVICE behind CAP_NET_RAW; a
// permission refusal is that gate, not a wiring defect.
Err(e) => assert!(format!("{e:#}").contains("SO_BINDTODEVICE lo"), "{e:#}"),
}
}
}
fn resolve_grpc_bind(spec: &BindSpec) -> Result<Option<SocketAddr>> {
match spec
.resolve_with(|name| {
appliance_pid1::net::interface_ipv4(name).map_err(|e| format!("{e:#}"))
})
.map_err(|why| anyhow::anyhow!("{why}"))?
{
gunnar_common::bindspec::Resolved::Bound(addr) => Ok(Some(addr)),
gunnar_common::bindspec::Resolved::NoAddress { interface, why } => {
// ERROR and not WARN: an operator who asked for a control plane
// has not got one, and nothing else on the box will say so.
tracing::error!(
interface = %interface,
"{why} The gRPC control plane will NOT start. The git listener is unaffected \
and this server still serves and authenticates pushes and fetches over SSH."
);
Ok(None)
}
}
}
/// `--admin-key` is either the key line itself or `@<path>`.
fn read_key_argument(spec: &str) -> Result<String> {
match spec.strip_prefix('@') {
Some(path) => std::fs::read_to_string(path)
.with_context(|| format!("reading the public key from {path}"))
.map(|text| text.trim().to_owned()),
None => Ok(spec.trim().to_owned()),
}
}
/// **The startup banner's host-key line.**
///
/// The fingerprint IN FULL, never shortened: the dashboard's `short_fp` is for
/// recognising a key you already know, and this line exists to be the first
/// place anybody learns it. It is read off a VNC screenshot by a human and
/// typed into a `known_hosts` line, and twelve characters cannot be typed into
/// anything.
///
/// The word `sshkey` leads it so a screenshot can be searched for one token.
///
/// Why this is worth a line of banner at all: `known_hosts` records no
/// provenance. Once written, a key taken off the wire by `ssh-keyscan` and a
/// key read off a console are the same bytes, for ever after, and nothing in
/// the file can tell the two apart. The only thing that makes a forge push
/// mean anything is WHERE the fingerprint came from — a record written by the
/// install (`deploy/upcloud`'s receipt) or a reading off the box's own console.
/// Never from whatever answered on `:2222`.
pub(crate) fn host_key_banner_line(fingerprint: &str) -> String {
format!("sshkey {fingerprint} (pin THIS; a key scanned off the port proves nothing)")
}
/// The host key, as russh's `ssh-key` sees it.
fn russh_private_key(openssh: &[u8]) -> Result<gunnar_ssh::russh::keys::PrivateKey> {
gunnar_ssh::russh::keys::PrivateKey::from_openssh(openssh)
.map_err(|err| anyhow::anyhow!("the stored host key did not parse: {err}"))
}
/// Run `gunnar serve` to completion.
pub async fn run(stores: Arc<ServerStores>, config: ServeConfig, check_only: bool) -> Result<()> {
// The descriptor limit, raised once. `ServerStores::open` already did this
// (its handle cache is sized from the answer); calling it here is the
// guarantee that the `limits.nofile` line — a WARN row on the dashboard
// when the limit is low — exists before a listener binds, whichever path
// built `stores`.
let _ = crate::limits::nofile();
// ── THE NARRATOR, BEFORE ANYTHING ELSE ──────────────────────────────────
//
// MEASURED 2026-09-23: `boot::ship_boot_narration` ends at `end of
// hand-off` — the moment the DHCP lease makes the wire usable — and the
// estate's twin died in the silence after it, having read its config,
// bound its listeners, taken its role, held its certificate and started
// following. None of that was said. From here on every boundary is.
//
// The sink is `korp.log_udp=` off the kernel command line and nothing
// else. Absent ⇒ nothing is armed, no thread is started, no datagram is
// sent, and the two file readers (the ESP's BOOTLOG.TXT and /data) carry
// the story as they always did. That is the off switch.
use crate::narrate::Stage;
crate::narrate::install();
crate::narrate::data_home(stores.root());
crate::narrate::at(
Stage::Config,
&format!(
"data root {}, engine {:?}, ssh {}, grpc {:?}, twin_role={}, is_twin={}",
stores.root().display(),
stores.engine(),
config.ssh_bind,
config.grpc_bind,
config.twin_role,
stores.is_twin(),
),
);
// ── PID 1's own duties, before anything binds ───────────────────────────
//
// The network first and by hand, because the interface has to exist before
// a listener can be reached over it and on this appliance nothing else
// brings it up: measured on the reference desktop 2026-08-03, the kernel is built with
// `CONFIG_IP_PNP` off, so `ip=` on the command line configures nothing.
// A refused static plan or malformed GUNNAR_NET is fatal on purpose: the
// alternative is a server that starts, logs "ssh 0.0.0.0:2222" and can
// never be reached, which is the same failure `Readiness::RefuseToServe`
// exists to prevent. A DHCP timeout is NOT fatal — bring_up_network warns
// and serves unreachable, the missing-GUNNAR_NET posture — because there
// the world declined to answer; the declaration did not lie.
crate::narrate::at(
Stage::Network,
"bringing the declared interfaces up over netlink",
);
let dhcp_session = crate::boot::bring_up_network()?;
// The lease. The UDP narration wire is retired (2026-09-24): the installer
// collects this box's whole boot log over its own mutual TLS.
crate::narrate::at(
Stage::Lease,
&format!("{} DHCP lease(s) held", dhcp_session.len()),
);
crate::narrate::at(
Stage::Sink,
"the boot log is kept for the installer (its bootstrap listener serves it)",
);
// **The clock, and here rather than anywhere later.** It is after the
// network because it is fetched over the network, and before every listener
// because `gunnar_grpc::auth::MAX_CLOCK_SKEW_MS` is ±5 minutes: an appliance
// that serves a single request on a clock two hours out refuses a genuine
// credential and a genuine Stripe webhook signature, which is precisely what
// gunnar.rs measured on 2026-09-14. Never fatal, never a `?`: see
// `crate::clock`'s note on why a failed clock fetch must not cost a forge.
crate::clock::synchronise_at_boot().await;
// And again when a credential is refused for skew, and every six hours: a
// boot-time step fixes the clock the box boots with, never the drift of the
// months until the next re-image. Background threads; see `crate::clock`.
crate::clock::arm_triggers();
// **The bind spelling becomes an address here, and not one line earlier.**
// `--grpc-bind eth1:50051` names an interface whose address arrives with
// the DHCP lease above; asking before `bring_up_network` would find an
// interface with no address every time. Asking AFTER is also what makes
// the refusal actionable: if it fires, the lease genuinely did not land,
// and the warning `bring_up_network` already logged says why.
let mut config = config;
// **A bind by ROLE** (`utility:50051`, the lean estate, A19 row 17) becomes this box's
// utility NIC here, after the network is up: `eth1` on a box with two NICs, `eth0` on a
// box laid behind its front with its utility NIC alone.
config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)
.context("resolving the --grpc-bind role to this box's NIC")?;
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
tracing::info!(interface = %interface, "the control plane's utility door is {interface} on this box");
}
// **An interface whose lease PID 1 still asks for** (its boot DHCP failed — estate-box3
// D4, a box rebooted while its network was down) is WAITED for, here on the appliance's
// path as in `Server::start`: the control plane binds once the address arrives.
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
let has_address = |i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
}
let resolved = resolve_grpc_bind(&config.grpc_bind).with_context(|| {
"resolving --grpc-bind after the network came up. A medium that names a NIC this \
machine does not have is a declaration that lies, and gunnar stops rather than \
falling back to a wildcard bind: an unauthenticated control plane on every \
interface is the defect the interface spelling exists to close."
})?;
let resolved_grpc_bind = match resolved {
Some(addr) => addr,
None => {
config.no_grpc = true;
gunnar_grpc::DEFAULT_BIND
}
};
// Pin the answer into the config so `Server::start`, the dashboard and the
// readiness line all report the address the listener took rather than the
// spelling it was asked for — and so the netlink question is asked once.
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
config.grpc_bind_device = Some(interface.clone());
}
config.grpc_bind = BindSpec::Addr(resolved_grpc_bind);
if !config.no_grpc && !resolved_grpc_bind.ip().is_loopback() {
tracing::info!(
grpc_bind = %resolved_grpc_bind,
"gRPC control plane will bind a NON-loopback address"
);
}
// Whether /data survived the last install (the ESP record korp-installer
// left) — one event, one dashboard row, before the band starts.
crate::boot::announce_install_mode(stores.root());
// The start-up disk test, appliance only (same gate as the dashboard, or
// `GUNNAR_DISKTEST=1`): one thread, one `disktest` row, never waited for.
// A32-14, stays a thread: the one-shot start-up disk test, appliance only; it writes its row and ends, never waited for.
crate::disktest::spawn(stores.root().to_path_buf());
if appliance_pid1::net::is_init() {
tracing::info!(
"running as pid 1: no init system underneath, so this process handles SIGTERM and \
reaps re-parented orphans itself"
);
}
// Installed unconditionally. A reaper that only exists at pid 1 is a reaper
// no test on a developer machine ever runs.
// A32-14, stays a task: the pid 1 reaper for the whole process; its handle is kept and aborted at the stop.
let reaper = tokio::spawn(appliance_pid1::runtime::reaper());
// **The soft-delete sweep**: the half of `repo delete` that gives the bytes
// back. Hourly, and once now. Without it a retention window would be a
// promise to keep repositories forever, and `/data` would fill with
// repositories nobody can list.
let tombstone_sweep = crate::stores::spawn_tombstone_sweep(Arc::clone(&stores));
// **The flow flush**: the half of the activity ledger that survives a
// restart. The counters a transfer bumps live in atomics in this process;
// this copies them into the control store every minute, and rolls the UTC
// day on a box that is idle at midnight. See `stores::spawn_flow_flush`.
// The band's fold, refreshed by the flush task once a minute. `None` until
// the first flush: a row of zeroes before anything has been folded would
// claim the box served nobody today.
let flow_band: Arc<std::sync::Mutex<Option<gunnar_core::Fleet>>> =
Arc::new(std::sync::Mutex::new(None));
let flow_flush = crate::stores::spawn_flow_flush(Arc::clone(&stores), Arc::clone(&flow_band));
config
.flow_view
.fill(move || *flow_band.lock().expect("the flow band"));
// Kept out of the move into `Server::start` below, for the shutdown flush.
let flow_at_exit = Arc::clone(stores.flow());
// The VGA status screen, appliance only (PID 1 booted by korp-start) or
// `GUNNAR_DASHBOARD=<device>`. One thread, fed by the log tap; the lease
// line above is already in the tap's buffer and is its first row. Never
// fatal, never waited for — see `crate::dashboard`.
// The band starts before the server does, so it is handed a VIEW and the
// replicator is dropped into it below, once `Server::start` has one. A box
// with no standby leaves it unfilled and the band draws no `standby` row.
let replication_view = config.replication_view.clone();
// A32-14, stays a thread: the VGA status screen, appliance only, fed by the log tap for the process's life; never fatal.
crate::dashboard::spawn(crate::dashboard::Facts {
ssh_bind: config.ssh_bind,
grpc_bind: (!config.no_grpc).then_some(resolved_grpc_bind),
data_root: stores.root().to_path_buf(),
build: gunnar_common::BUILD_REF.to_owned(),
engine: Some(stores.engine()),
replication: replication_view.clone(),
tenants: config.tenants_view.clone(),
flow: config.flow_view.clone(),
});
// Exactly when the address is DHCP-learned, a task keeps the lease alive
// (RFC 2131 §4.4.5 — appliance_pid1::runtime::dhcp_renewal). Static and unconfigured paths
// return an EMPTY list above and gain no task: renewal exists only where a
// lease does — and a two-NIC appliance holds TWO leases, each with its own
// T1, so there is one task per lease rather than one task guessing which
// interface it is keeping alive.
#[cfg(feature = "dhcp-client")]
let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
.into_iter()
// A32-14, stays a task: one lease's renewal for the process's life; the handles are kept and aborted at the stop.
.map(|session| tokio::spawn(crate::boot::dhcp_renewal(session)))
.collect();
// Built without `dhcp-client`: `DhcpSession` is UNINHABITED, so this is the
// compiler proving there is no lease rather than a runtime check claiming
// it. The `Some` arm is `match session {}` — unreachable by construction —
// and the abort sites below need no cfg of their own.
#[cfg(not(feature = "dhcp-client"))]
let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
.into_iter()
.map(|session| match session {})
.collect();
// ── the tunnel, AFTER the outer address ─────────────────────────────────
//
// Order is load-bearing and not alphabetical: a WireGuard peer is reached
// over the outer address, so a handshake attempted before DHCP has
// finished has no source address to leave from. It is strictly additive —
// no `wg0.conf` in the data root means one INFO line and nothing else
// changes — and it is deliberately NOT fatal when the kernel refuses,
// because an appliance that will not serve git because a tunnel failed is
// worse than one that serves on its outer address and says so loudly.
//
// Nothing below has to be told about it: both listeners bind `0.0.0.0`
// (`ServeConfig::default`, ssh 0.0.0.0:2222), so `wg0` is served the
// moment it has an address.
// **This binding is the tunnel's lifetime.** The data plane is a thread in
// THIS process now, not a device the kernel owns: `Tunnel` holds the pump
// and the tun fd, and dropping the last handle stops the pump and deletes
// `wg0`. The clone below goes to a watcher task that RETURNS after the
// first handshake, so if this binding were consumed instead of kept the
// tunnel would come up, announce itself and vanish a few seconds later.
// It stays in scope for the whole of `run`, which is the process.
// A pairing box's `wg0.conf` is written by the pairing loop once the
// installer's door has assigned the address (pairing.rs); on a reboot it
// is already here and this lane raises it like any other.
#[cfg(feature = "wg-client")]
let tunnel = crate::wg::bring_up(stores.root())?;
#[cfg(feature = "wg-client")]
let handshake = tunnel.clone().map(|t| {
// A32-14, stays a task: the tunnel's first-handshake watcher; it returns after the handshake, its handle is aborted at the stop.
tokio::spawn(crate::wg::handshake_watch(
t,
20,
std::time::Duration::from_secs(3),
))
});
// Built without `wg-client`: there is no tunnel type, no boringtun and no
// `wg_appliance_core` in this binary at all, so there is nothing to bring
// up, nothing to watch and no banner line to write. No `tunnel` binding
// exists in this build — only the watcher handle the two abort paths below
// reach for, and it is permanently `None`.
#[cfg(not(feature = "wg-client"))]
let handshake: Option<tokio::task::JoinHandle<()>> = None;
// Read off both BEFORE the move below, so the role line can be narrated
// after the listeners are up: `Server::start` consumes `stores` and
// `config`, and the role is a property of the boot rather than of the
// server object.
let config_twin_role = config.twin_role;
let stores_is_twin = stores.is_twin();
// `Server::start` fills `config.replication_view` — the same object the
// band above was handed — once the replicator exists.
let server = Server::start(stores, config).await?;
// The staleness alarm: one WARN into the log and one event-stream event the first
// time a standby stops taking bytes — with "last successful replication:
// NEVER" as its own message. Harmless when nothing is configured: the view
// reports no standbys and the task never fires.
let staleness_watch = crate::replication_view::spawn_staleness_watch(
replication_view.clone(),
Arc::clone(server.bus()),
);
// The banner is BUILT once and written twice (one implementation, one writer): to stdout,
// which on the appliance is `/dev/console` = the serial line that
// `installtest/src/iso.rs` reads its boot proof off, and — appliance only —
// mirrored to `/dev/tty0` so a KVM GRAPHICS console ends on gunnar's own
// status instead of the kernel's last printk. Measured 2026-08-12: with the
// banner on serial alone, a VGA console showed a healthy boot log that never
// once said "gunnar".
let mut banner = String::new();
{
use std::fmt::Write as _;
writeln!(banner, "ssh {}", server.ssh_addr()).ok();
// **THE HOST KEY, ON A CHANNEL PRODUCTION HAS.** Directly under the
// address it belongs to, because the two together are the whole of
// what a client needs to trust this box.
//
// MEASURED 2026-09-23. The fingerprint was logged and nowhere else:
// `tracing` → stderr → `/dev/console` → the LAST `console=` on the
// command line → the serial line, and the UpCloud shape has no serial
// line (a panic there prints PID 1's first `print!` to nothing; the
// ISO's own gate VMs are the only boxes that ever had it, in their
// `serial.log`). So in production the one fact a client must pin
// reached NOBODY, and this laptop's `known_hosts` has eight
// `ssh-keyscan`-shaped lines for `[git.gunnar.rs]:2222` to prove what
// fills the gap.
//
// This banner is the channel that answers: it is printed on stdout AND
// mirrored to `/dev/tty0` (`appliance_pid1::console::mirror_to_gfx_console`, a few
// lines below), which is the VNC screen of an UpCloud VM. The
// WireGuard public key is already here for exactly this argument — on
// a shell-less appliance the banner is the only channel that can
// answer "is the key on this box the one I enrolled?".
//
// The narration wire (`korp.log_udp=`) was considered FIRST and is
// not usable: it is armed and flushed long before this point, but the
// one door onto it (`appliance_pid1::narration::narration_door`) classifies any
// line carrying `sha256:` as `pubkey` and replaces it with a withheld
// notice, the collector refuses it identically
// (`<ops>/xtask/src/bootnarr.rs::secret_class`), and the 40-character
// opaque-run rule catches the fingerprint's body with the label
// stripped off — so no spelling escapes. That door is an OWNER RULE
// ("a PUBLIC verifying key … is withheld anyway, because it is not
// progress"), spelled in three repositories. Putting the fingerprint
// on that wire needs the rule reversed and all three vocabularies
// changed together; it is not something this line may do quietly.
writeln!(
banner,
"{}",
host_key_banner_line(server.host_key_fingerprint())
)
.ok();
match server.grpc_endpoint() {
// The open enrolment window comes FIRST on this line, before the
// reassuring "authenticated": a server that has never been enrolled is
// one anybody who can reach the port can take, and an operator reading
// three lines of banner must not have to know that "authenticated"
// does not cover `Bootstrap.Enroll`.
Some((addr, _)) if server.bootstrap_window_open() => writeln!(
banner,
"grpc {addr} (NOT ENROLLED — Bootstrap.Enroll is open to anyone who can reach \
this port. `gunnar control bootstrap <name>` claims it, once.)"
)
.ok(),
Some((addr, true)) => writeln!(
banner,
"grpc {addr} (authenticated: sshsig + the ACL model)"
)
.ok(),
Some((addr, false)) => writeln!(
banner,
"grpc {addr} (UNAUTHENTICATED — the escape hatch is on)"
)
.ok(),
None => writeln!(banner, "grpc disabled").ok(),
};
// Only when it is on, like metrics below. Without this line a server asked
// to bind HTTP on port 0 has bound a listener nobody can find: ssh, grpc,
// metrics and the log level were all announced, and the one listener a
// stock `git` client talks to was not.
#[cfg(feature = "http")]
if let Some(addr) = server.http_addr() {
if server.http_tls() {
writeln!(
banner,
"https {addr} (TLS terminated here; clients pin the issuer)"
)
.ok();
} else {
writeln!(banner, "http {addr}").ok();
}
}
// Only when it is on: a line reading "metrics disabled" on every
// server that never asked for the feature is banner an operator learns to
// skip, and the banner's whole value is that it is short enough to read.
if let Some(addr) = server.metrics_addr() {
writeln!(
banner,
"metrics http://{addr}{} (no authentication{})",
crate::scrape::SCRAPE_PATH,
if addr.ip().is_loopback() {
""
} else {
"; NOT on loopback"
}
)
.ok();
}
// The tunnel, and ONLY when there is one — same rule as metrics above.
//
// It carries the appliance's WireGuard PUBLIC KEY, which is not
// decoration: peering is mutual allowlisting by public key, and a key
// the hub does not have in its `[Peer]` stanza gets **silence** rather
// than a rejection. So "is the key on this box the one I enrolled?" is
// the first question of every failed tunnel, and on a shell-less
// appliance the banner — serial AND the mirrored graphics console — is
// the only channel that can answer it.
#[cfg(feature = "wg-client")]
if let Some(tunnel) = &tunnel {
writeln!(banner, "{}", tunnel.banner_line()).ok();
}
// On stdout beside the addresses rather than in the log: this is
// the one fact an operator needs BEFORE the incident, and by the time they
// want it the log is the thing they cannot read enough of. It also warns
// them off the trap — both signals default to terminating a process, and
// that is what they did to this one until today.
writeln!(
banner,
"log {} (SIGUSR1 → one rung more verbose, SIGUSR2 → back; pid {})",
gunnar_common::logging::current().unwrap_or_else(|| "?".to_owned()),
std::process::id()
)
.ok();
}
print!("{banner}");
appliance_pid1::console::mirror_to_gfx_console(&banner);
// **ONE LINE PER LISTENER, WITH THE ADDRESS IT TOOK.** Not the spelling it
// was asked for: `--grpc-bind eth1:50051` and `10.13.9.24:50051` are two
// different facts and only the second one can be dialled. A listener the
// build does not carry says nothing rather than saying `disabled`, so a
// reader counting listeners counts the ones that exist.
crate::narrate::at(Stage::Listener, &format!("ssh BOUND {}", server.ssh_addr()));
match server.grpc_endpoint() {
Some((addr, _)) => crate::narrate::at(
Stage::Listener,
&format!(
"grpc BOUND {addr}{}",
if server.bootstrap_window_open() {
" (NOT ENROLLED — Bootstrap.Enroll is open)"
} else {
""
}
),
),
None => crate::narrate::at(Stage::Listener, "grpc NOT bound — the control plane is off"),
}
#[cfg(feature = "http")]
if let Some(addr) = server.http_addr() {
crate::narrate::at(
Stage::Listener,
&format!(
"{} BOUND {addr}",
if server.http_tls() { "https" } else { "http" }
),
);
}
if let Some(addr) = server.metrics_addr() {
crate::narrate::at(Stage::Listener, &format!("metrics BOUND {addr}"));
}
// **THE ROLE**, from the two places that decide it: what the medium
// declared and what the data home already is. They are narrated together
// because a disagreement between them is the interesting case — a box
// installed from main's medium onto a disk that already holds a twin's
// certificate is exactly the shape the estate re-images into.
crate::narrate::at(
Stage::Role,
&format!(
"{} (declared twin_role={}, data home says is_twin={})",
if stores_is_twin || config_twin_role {
"TWIN"
} else {
"PRIMARY"
},
config_twin_role,
stores_is_twin,
),
);
if check_only {
reaper.abort();
tombstone_sweep.abort();
flow_flush.abort();
staleness_watch.abort();
for renewal in &renewals {
renewal.abort();
}
if let Some(handshake) = &handshake {
handshake.abort();
}
server.shutdown().await?;
return Ok(());
}
// **SERVING.** The end of the boot story and the beginning of the ticks:
// from here the only thing the box is expected to say is one short line a
// minute, and the ear's threshold is measured against that. A box that
// reached here and then went quiet is a DEATH AT A NAMED PLACE — which is
// the reading the estate did not have on 2026-09-23.
crate::narrate::at(
Stage::Serving,
&format!("up at {} s", crate::narrate::uptime_s(),),
);
let result = server.wait().await;
reaper.abort();
tombstone_sweep.abort();
// One last flush before the process goes. The flush COPIES rather than
// adds, so doing it here as well as on the tick cannot double a count, and
// it is what stops an orderly shutdown losing the last minute of today.
if let Err(err) = flow_at_exit.flush(gunnar_core::now_ms()) {
tracing::warn!(
target: "gunnar::store",
event = "flow.flush_failed",
%err,
"the flow ledger could not be written on shutdown; up to one flush interval of \
today's activity counters is lost"
);
}
flow_flush.abort();
staleness_watch.abort();
for renewal in &renewals {
renewal.abort();
}
if let Some(handshake) = &handshake {
handshake.abort();
}
result
}
/// The public key argument reader, exposed for the wiring tests.
#[cfg(test)]
mod tests {
/// GUN-PERF: every connection a gRPC door accepts has `TCP_NODELAY` set, so a small
/// answer is not held back by Nagle until the client's delayed ACK (about 40 ms).
#[tokio::test]
async fn a_grpc_door_accepts_its_connections_with_nodelay() {
use tokio_stream::StreamExt as _;
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind");
let at = listener.local_addr().expect("addr");
let mut incoming = Box::pin(super::grpc_incoming(listener));
let _client = tokio::net::TcpStream::connect(at).await.expect("connect");
let accepted = incoming
.next()
.await
.expect("one connection")
.expect("accepted");
assert!(
accepted.nodelay().expect("read TCP_NODELAY"),
"a gRPC door accepted a connection without TCP_NODELAY"
);
}
use super::read_key_argument;
use super::*;
/// **The banner line an operator reads off the VNC screen.**
///
/// The fingerprint must be there IN FULL and unmangled: this line is the
/// only production channel that carries it (the log goes to a serial
/// console the UpCloud shape does not have, and the narration wire's one
/// door withholds anything containing `sha256:`), and a fingerprint that
/// has been shortened for display cannot be pinned by anybody.
///
/// Red-first: shortening the fingerprint the way the dashboard's
/// `short_fp` does, or dropping it from the line, fails here.
#[test]
fn the_banner_names_the_host_key_in_full() {
const FP: &str = "SHA256:z/fSv0Z0RZS+Lccbc6ZoOWwt/fbj1VFJGSBKQbo4icE";
let line = host_key_banner_line(FP);
assert!(line.contains(FP), "{line}");
assert!(line.starts_with("sshkey "), "one searchable token: {line}");
assert!(
!line.contains('\n'),
"one line, or the banner is not a banner: {line}"
);
// And it warns off the thing that fills known_hosts when this line is
// missing, because a fingerprint with no provenance is not evidence.
assert!(
line.contains("scanned off the port proves nothing"),
"{line}"
);
}
/// The visible limit has to sit **inside** the pool that carries
/// it, or the invisible ceiling is back, at whichever of the two numbers is
/// smaller, which is the one nobody is looking at.
#[test]
fn the_blocking_pool_is_larger_than_the_transfer_limit() {
let transfers = gunnar_ssh::SshConfig::default()
.max_concurrent_transfers
.expect("the appliance default must have a transfer limit");
assert!(
transfers < GIT_BLOCKING_THREADS,
"max_concurrent_transfers ({transfers}) is not below GIT_BLOCKING_THREADS \
({GIT_BLOCKING_THREADS}), so the blocking pool becomes the real ceiling again and \
the admission counters would show a server that is never full while every transfer \
queues"
);
assert!(
GIT_BLOCKING_THREADS >= transfers * 2,
"the blocking pool ({GIT_BLOCKING_THREADS}) leaves no headroom over \
{transfers} concurrent transfers for the other blocking work a transfer needs: \
pack indexing, ref transactions, the arrow log"
);
}
/// The control plane must get a pool that is bounded on both sides — and
/// at least FOUR workers (R5.1-G): two slow synchronous calls (a snapshot,
/// a disk walk) held both workers of the old floor, and every other call —
/// the console's page loads, monetize's `Entitlement.Set` — waited behind
/// them.
#[test]
fn the_control_plane_gets_its_own_bounded_pool() {
let workers = control_plane_workers();
assert!(
(4..=8).contains(&workers),
"control_plane_workers() returned {workers}; below four, two slow synchronous calls \
hold every worker and the whole administrative surface queues behind them, and an \
unbounded pool recreates the starved listener with extra steps"
);
}
#[test]
fn a_literal_key_and_an_at_path_read_the_same() {
let dir = tempfile::tempdir().expect("a temp dir");
let path = dir.path().join("id.pub");
std::fs::write(&path, "ssh-ed25519 AAAAC3Nz literal\n").expect("write");
assert_eq!(
read_key_argument("ssh-ed25519 AAAAC3Nz literal").expect("literal"),
read_key_argument(&format!("@{}", path.display())).expect("from file"),
"`--admin-key @file` and `--admin-key '<line>'` must produce the same key, or a \
bootstrap that works one way fails the other"
);
}
/// The 32-CPUs-of-affinity-on-8-CPUs-of-quota condition, seen **both** ways.
///
/// A warning that reads its own host can only ever be observed in the state
/// that host is in, and a build box that happens to be unconstrained would
/// have exercised the silent branch for ever while the loud one rotted.
#[test]
fn a_quota_far_narrower_than_the_affinity_is_reported_and_a_matching_one_is_not() {
// The measured bad case: `podman run --cpus 8` on a 32-core host.
assert_eq!(
spread_budget(Some(32), Some(8.0)),
Some((32, 8)),
"eight CPUs of quota spread over thirty-two is the condition that cost 2.7x the \
CPU per request at 32 clients; it must be said out loud"
);
// The same budget as a cpuset: nothing to report.
assert_eq!(spread_budget(Some(8), Some(8.0)), None);
// Exactly at the line, and one either side of it, because `>=` and `>`
// are one character apart and both compile.
assert_eq!(spread_budget(Some(16), Some(8.0)), Some((16, 8)));
assert_eq!(spread_budget(Some(15), Some(8.0)), None);
// A fractional quota rounds UP to the CPUs an operator would name.
assert_eq!(spread_budget(Some(32), Some(1.5)), Some((32, 2)));
// No quota at all is the ordinary case and is not a finding. Neither is
// an unreadable affinity: half a comparison would be a number invented
// to fill the other half.
assert_eq!(spread_budget(Some(32), None), None);
assert_eq!(spread_budget(None, Some(8.0)), None);
}
/// `Cpus_allowed_list` is ranges *and* singletons, and the two mixed.
///
/// Against literals rather than against this host: the reference desktop reports a plain
/// `0-31`, and a parser that counted ranges and silently dropped singletons
/// passed on this box while answering 0 for a server pinned to one CPU.
#[test]
fn a_cpu_list_is_counted_whether_it_is_ranges_singletons_or_both() {
assert_eq!(count_cpu_list("0-31"), Some(32));
assert_eq!(count_cpu_list("0-7"), Some(8));
assert_eq!(
count_cpu_list("3"),
Some(1),
"one pinned CPU is one CPU, not none"
);
assert_eq!(count_cpu_list("0,2,4,6"), Some(4));
assert_eq!(count_cpu_list("0-3,8,10-11"), Some(7));
assert_eq!(
count_cpu_list(" 0-3 , 8 \n"),
Some(5),
"the kernel line is padded"
);
// Unreadable is None, never a number: a guessed count is the one thing
// that turns this whole check into a lie.
assert_eq!(count_cpu_list(""), None);
assert_eq!(count_cpu_list("banana"), None);
assert_eq!(
count_cpu_list("7-3"),
None,
"a reversed range is not 5 CPUs"
);
}
#[test]
fn the_affinity_of_this_process_is_read_as_a_count_of_cpus() {
let n = cpus_allowed().expect("Linux always publishes Cpus_allowed_list");
assert!(
(1..=1 << 20).contains(&n),
"this process reports {n} allowed CPUs, which is not a count of CPUs"
);
// The whole comparison rests on this being the *affinity* and not the
// machine, so it must never exceed what the machine has.
let online = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
assert!(
n >= online,
"the affinity ({n}) is narrower than what std reports as available ({online}), \
which would make every comparison against it too quiet"
);
}
#[test]
fn a_missing_key_file_is_an_error_not_an_empty_key() {
let err = read_key_argument("@/nonexistent/gunnar/key.pub")
.expect_err("a missing key file must fail");
// An empty string here would reach `bootstrap_admin` and fail there
// with a parse error, but it would also be indistinguishable from "no
// key was asked for" in any caller that checked for emptiness.
assert!(format!("{err:#}").contains("reading the public key"));
}
}
/// **The connections a gRPC door accepts, each with `TCP_NODELAY`** (lane GUN-PERF, 2026-10-31).
///
/// tonic sets `TCP_NODELAY` only on a listener it binds itself; with
/// `serve_with_incoming*` the setting is ignored (tonic 0.14 says so on the method), and
/// every door here hands tonic its own listener. Without it a small answer's last
/// write (the trailers frame after the data frame) waits for the client's delayed ACK:
/// about 40 ms per call on Linux, which the console pays on every page it builds from
/// gunnar's answers. The gRPC twin of `gunnar-http`'s and `gunnar-ssh`'s accept loops.
pub(crate) fn grpc_incoming(
listener: tokio::net::TcpListener,
) -> impl tokio_stream::Stream<Item = std::io::Result<tokio::net::TcpStream>> {
use tokio_stream::StreamExt as _;
tokio_stream::wrappers::TcpListenerStream::new(listener).map(|accepted| {
The file goes on; the console shows all of it.