gunnar_server/serve.rs
1//! `gunnar serve` starts its listeners together and stops them together. They
2//! share nothing but the control plane behind them.
3//!
4//! ```text
5//! :2222 SSH git-upload-pack / git-receive-pack, publickey auth
6//! :8080 HTTP the same two verbs over smart HTTP, access-token
7//! auth — only with `--features http` AND `--http-bind`
8//! 127.0.0.1:50051 gRPC the control plane; authenticated (sshsig + the
9//! ACL model) unless the development escape hatch is on
10//! ```
11//!
12//! **The two git listeners are two doors onto one room**, and the wiring below
13//! is what keeps that true: the same `Dispatch`, the same `WireTransport`, the
14//! same `ServerMetrics` and the one `allow_anonymous` field are handed to both.
15//! A second control plane or a second transport here is how SSH and HTTP would
16//! come to disagree about who may push.
17//!
18//! Git traffic never passes through gRPC. There is no bridge and none is
19//! wanted: it is that separation which makes the v1 authentication hole
20//! survivable, because an open gRPC port cannot push code.
21//!
22//! # What this module is responsible for saying out loud
23//!
24//! 1. **Whether the unauthenticated enrolment door is open**, at every start,
25//! on stdout *and* in the log. `gunnar_core::AuthTable::prepare_to_serve`
26//! returns the state as a value; [`announce_readiness`] is the caller that
27//! has to branch on it, and `tests/bootstrap.rs` asserts every branch.
28//!
29//! Until 2026-08-05 an empty identity table with no `--admin-key` was a
30//! **refusal to start** (the secret-store seam). That is why gunnar had no first boot: the
31//! box could not be reached to be given its first key, and the only way in
32//! was `gunnar key add` with the server down, which needs a shell the
33//! appliance does not have. It now starts and serves `Bootstrap.Enroll`
34//! until somebody uses it — see `gunnar_grpc::service::bootstrap`.
35//! 2. **gRPC binds loopback unless told otherwise**, and says so at startup
36//! whatever it binds.
37//!
38//! # Two listeners, and now two runtimes
39//!
40//! Separate listeners separate the *ports*. They did not separate the *schedulers*, and that
41//! turned out to be most of what the separation was for.
42//!
43//! `ControlPlane` is a synchronous trait — `gunnar-grpc`'s `seam.rs` says so,
44//! and says the backend belongs behind `spawn_blocking` "at the call site" —
45//! and no call site does it. `Admin.StorageInfo` runs an unbounded recursive
46//! `std::fs::read_dir` that `stat`s every loose object in the repository,
47//! straight from its `async fn`, on a worker thread. `Repos.Create` opens a git
48//! directory and replays an arrow log. While both listeners shared one runtime,
49//! a handful of concurrent control-plane calls occupied every worker and the SSH
50//! listener's accept, handshake and exec dispatch simply did not run. The loopback default leaves
51//! the control port unauthenticated, so that was reachable on purpose as well as
52//! by accident.
53//!
54//! **The control plane therefore gets its own runtime**, built here, with its
55//! own worker threads and its own blocking pool. Two properties follow that
56//! thirty `spawn_blocking` wrappers would not have given:
57//!
58//! * it holds for *every* control-plane call, including ones written later by
59//! somebody who never read `seam.rs`. The fix is by construction rather than a
60//! convention thirty call sites have to keep (one implementation);
61//! * a control plane that stalls completely degrades the control plane only. Git
62//! keeps being served.
63//!
64//! The gRPC listener is bound **on** that runtime rather than handed to it: a
65//! tokio socket registers with the I/O driver of the runtime that created it,
66//! and one whose driver belongs to a runtime that is shutting down is a stall
67//! waiting to happen. `tests/control_plane_starvation.rs` measures the whole
68//! thing from an OS thread outside both runtimes.
69
70use std::net::SocketAddr;
71use std::sync::Arc;
72
73use anyhow::{bail, Context, Result};
74use gunnar_core::observe::{MetricsSnapshot, RequestId, ServerMetrics};
75use gunnar_core::{ensure_host_key, Readiness};
76use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
77use gunnar_ssh::{BoundServer, GunnarSsh, SshConfig};
78use tokio::sync::oneshot;
79use tokio::task::JoinHandle;
80
81use gunnar_common::bindspec::BindSpec;
82
83use crate::control_plane::CoreControlPlane;
84use crate::stores::ServerStores;
85use crate::transport::{WireConfig, WireTransport};
86
87/// What `gunnar serve` was asked for.
88#[derive(Debug, Clone)]
89pub struct ServeConfig {
90 /// Where the SSH listener binds.
91 pub ssh_bind: SocketAddr,
92 /// Where the gRPC control plane binds. Loopback by default.
93 ///
94 /// A [`BindSpec`] rather than a `SocketAddr` because an appliance's private
95 /// address is DHCP-learned: the declaration names `eth1:50051` and
96 /// [`run`] resolves it, ONCE, after `GUNNAR_NET` has brought the link up
97 /// and before anything binds. Downstream — the readiness line, the
98 /// dashboard row, `Admin.Health` — then reports the address the listener
99 /// actually took, not the spelling it was asked for.
100 pub grpc_bind: BindSpec,
101 /// The DEVICE the control plane listens on, when `grpc_bind` was spelled as
102 /// an interface (`eth1:50051`) — set by [`run`]/[`start`] from that spelling
103 /// and honoured as `SO_BINDTODEVICE` on the listener socket, so the
104 /// interface name means the interface and not merely the address it held
105 /// at boot. MEASURED 2026-09-05 (the KVM bench rig, two slirp NICs on the
106 /// same 10.0.2.0/24): an address-only bind of eth1's 10.0.2.15 answered
107 /// through eth0, because both NICs held that address and the route decided.
108 /// On a cloud whose utility NIC is `eth1` on its own subnet the two agree;
109 /// the device bind makes them agree everywhere. `None` for an address spec.
110 pub grpc_bind_device: Option<String>,
111 /// An OpenSSH public key line to install as the bootstrap admin when the
112 /// identity table is empty (the secret-store seam).
113 pub admin_key: Option<String>,
114 /// **The keys `Bootstrap.Restore` is taken from** (`--restore-pubkey`, h-v1-6a): OpenSSH
115 /// public key lines or `@<path>`, resolved and checked at start. An installer-fed box adds
116 /// the fed admin keys itself; a box with neither refuses every restore by name.
117 pub restore_pubkeys: Vec<String>,
118 /// Start the SSH listener only.
119 pub no_grpc: bool,
120 /// **`--wg-only`: every box-to-box connection runs over the estate's
121 /// WireGuard overlay** (owner, 2026-10-04). The address the control plane
122 /// is bound on (`eth1:50051`, the utility network) serves NOTHING: the
123 /// control plane and the replication listener answer only on the tunnel's
124 /// address ([`crate::tunnel_addr`]), and SSH — bound for the public's git
125 /// clients — drops a connection that arrives at the utility address before
126 /// a byte is spoken. A plain-IP connection between two boxes is refused.
127 pub wg_only: bool,
128 /// `--role twin`: no forge listener from the first boot (loopback only),
129 /// whether or not `Bootstrap.BecomeTwin` has run yet.
130 pub twin_role: bool,
131 /// Whether a caller with **no key** may clone a repository marked
132 /// [`Visibility::Public`](gunnar_core::Visibility::Public). `true` by
133 /// default; nothing is anonymously readable until a repository is
134 /// deliberately made public, because `Private` is the default for a new
135 /// one.
136 ///
137 /// Reaches [`SshConfig::allow_anonymous`](gunnar_ssh::SshConfig::allow_anonymous)
138 /// **and `HttpConfig::allow_anonymous`**, from this one field. Two fields
139 /// is how one of them ends up stubbed: an operator who passed
140 /// `--no-anonymous` and got it honoured on one transport has a server that
141 /// serves strangers and reports that it does not. The gRPC control plane
142 /// has no anonymous mode to switch, at any setting.
143 pub allow_anonymous: bool,
144 /// **`--anonymous allow|refuse`** (the browse replica): whether an
145 /// anonymous read is SERVED here or sent to the browse replica the
146 /// refusal names. Orthogonal to `allow_anonymous`, which is whether a
147 /// stranger authenticates at all; this is what the dispatcher tells one
148 /// who did. Both front doors read it through the one `Control`.
149 pub anonymous: gunnar_ssh::AnonymousPolicy,
150 /// Where the smart-HTTP listener binds. `None` does not start
151 /// it, which is the default.
152 ///
153 /// Off unless asked for, because a second listening port is a decision an
154 /// operator makes and not one a default makes for them. The feature it
155 /// needs is off by default too, so an appliance image compiles no HTTP at
156 /// all.
157 #[cfg(feature = "http")]
158 pub http_bind: Option<SocketAddr>,
159 /// Whether the HTTP listener will read a credential off a plain connection
160 /// Default: only when a proxy attests TLS.
161 #[cfg(feature = "http")]
162 pub http_cleartext: gunnar_http::Cleartext,
163 /// Terminate TLS on the smart-HTTP listener itself. `None` — the default —
164 /// is the proxy-terminated shape; `SelfMinted` is the single-executable
165 /// shape (leaf minted from this server's own issuer at start, chain
166 /// closed for clients that pin the issuer); `Identity` is
167 /// bring-your-own-PEM.
168 #[cfg(feature = "http")]
169 pub http_tls: Option<HttpTls>,
170 // (HttpTls is defined right below ServeConfig.)
171 /// **Build every push**, in this process, hermetically (roadmap phase 3).
172 ///
173 /// `None` — the default — is a server that stores and serves and does not
174 /// build. `Some` starts the runner on the same event bus every push
175 /// publishes on, which is what makes the golden path's fourth step
176 /// *nothing at all*.
177 ///
178 /// It is in the serving process on purpose: a second process cannot
179 /// subscribe to an in-process bus, and a runner that polled the attestation
180 /// log instead would build a push some time *after* it landed rather than
181 /// *because* it landed.
182 pub runner: Option<crate::runner::WatchConfig>,
183 /// Require a **client certificate** on the gRPC control plane, minted by
184 /// this server's own issuer (the mTLS plan).
185 ///
186 /// `None` — the default — is the loopback shape the unauthenticated v1 assumes: the listener
187 /// binds `127.0.0.1` and a credential in metadata is the identity.
188 /// `Some(GrpcTls::SelfMinted)` is the shape a control plane reachable off
189 /// the box needs, and it changes two things at once: the transport refuses
190 /// a peer whose certificate this issuer did not sign, and the gate maps the
191 /// certificate that got through to an ACL principal.
192 ///
193 /// Deliberately **not** on by default. Turning it on rewrites how every
194 /// existing client must connect, and a security default that silently
195 /// breaks the working configuration gets switched back off in a hurry —
196 /// which leaves it off everywhere. It is a flag an operator turns on when
197 /// the listener leaves loopback, and `--grpc-bind` off loopback without it
198 /// is what the startup warning names.
199 pub grpc_tls: Option<GrpcTls>,
200 /// `--web-relying-party`, unparsed: the WebAuthn relying parties the gRPC
201 /// plane answers for (`gunnar_grpc::webauthn::RelyingParty`).
202 pub web_relying_parties: Vec<String>,
203 /// `--web-require-user-verification` (A34-12).
204 pub web_require_user_verification: bool,
205 /// `--identity-root-site` (IDENTITY-HOST): sign-up is open here.
206 pub identity_root_site: bool,
207 /// **The mutual replication listener** (`--grpc-replication-bind`, lane T9):
208 /// the door a twin follows this box through. Served only with a client
209 /// certificate this box's issuer signed, TLS 1.3 only, its leaf minted from
210 /// the issuer at every start under the fixed name
211 /// `gunnar_grpc::client_signer::REPLICATION_SERVER_NAME`.
212 ///
213 /// `None` DERIVES it, and the derivation is what lets the ONE medium carry
214 /// no flag: a control plane bound off loopback (the appliance,
215 /// `0.0.0.0:50051`) gets this listener on the same address, port + 1
216 /// (`50052`); a loopback one (a laptop, a test) gets none. Its main
217 /// listener stays exactly what it was, because turning `--grpc-tls-self` on
218 /// for the whole plane would lock out every client that has no certificate
219 /// — and the bootstrap door with them.
220 pub grpc_replication_bind: Option<SocketAddr>,
221 // (HttpTls is defined right below ServeConfig.)
222 /// How many git transfers may be in flight across the whole server.
223 /// `None` takes
224 /// [`DEFAULT_MAX_CONCURRENT_TRANSFERS`](gunnar_ssh::config::DEFAULT_MAX_CONCURRENT_TRANSFERS).
225 ///
226 /// The fix note on the admission gate says every one of these numbers needs to be
227 /// settable, because the right one is deployment-specific; until now
228 /// neither was reachable from `gunnar serve` and the two defaults were the
229 /// only values any deployment could have. That is also why the pressure
230 /// harness could not reach contention without 320 client processes.
231 pub max_concurrent_transfers: Option<usize>,
232 /// The **memory budget** the admission ledger may give out to git
233 /// operations in flight, in bytes (gate 2). `None` takes
234 /// [`crate::budget::budgets`]'s derived figure — `MemAvailable` x
235 /// [`crate::budget::TRANSFER_MEMORY_PERCENT`] — rather than
236 /// `gunnar-ssh`'s compiled-in 8 GiB, which was a number about the cgroup
237 /// one bakeoff ran in and not about this box.
238 pub memory_budget_bytes: Option<u64>,
239 /// How long a transfer may **wait** for a slot before it is refused.
240 /// `None` takes [`gunnar_ssh::governor::DEFAULT_ADMISSION_WAIT`];
241 /// `Some(Duration::ZERO)` restores the instant refusal.
242 pub admission_wait: Option<std::time::Duration>,
243 /// A principal's guaranteed floor and its share of a contended server.
244 /// `None` takes
245 /// [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`](gunnar_ssh::config::DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL).
246 pub max_transfers_per_principal: Option<usize>,
247 /// How many git reads may be served **without a key** at once.
248 /// `None` takes
249 /// [`DEFAULT_MAX_ANONYMOUS_TRANSFERS`](gunnar_ssh::admission::DEFAULT_MAX_ANONYMOUS_TRANSFERS).
250 ///
251 /// Note the asymmetry with the field above, which is deliberate: that one
252 /// is a floor a principal is guaranteed and may exceed on a quiet box, this
253 /// one is a wall that binds however empty the server is. See
254 /// [`gunnar_ssh::admission`] for why work-conserving is the right instinct
255 /// for a tenant and the wrong one for a stranger.
256 pub max_anonymous_transfers: Option<usize>,
257 /// **The usage share** (`admit-share`): `--admit-limit` and
258 /// `--admit-window-days`, parsed. See [`gunnar_ssh::SshConfig::admit`].
259 #[cfg(feature = "admit-share")]
260 pub admit: gunnar_ssh::edda_admit::Settings,
261 /// Make the per-principal number an unconditional wall again rather than a
262 /// max-min share. See
263 /// [`SshConfig::per_principal_limit_is_hard`](gunnar_ssh::SshConfig::per_principal_limit_is_hard).
264 pub per_principal_limit_is_hard: bool,
265 /// How many worker threads the control plane's **own** runtime gets.
266 /// `None` derives it from [`control_plane_workers`].
267 ///
268 /// A knob rather than a constant, because the right number is a deployment
269 /// question: a box whose operators live in the UI wants more, an appliance
270 /// serving one CI fleet wants fewer. A test also has to be able to make the
271 /// pool small enough to saturate it deliberately.
272 pub control_plane_workers: Option<usize>,
273 /// The largest object `Refs.ReadObject` will inflate. `None`
274 /// takes
275 /// [`CoreControlPlane`](crate::CoreControlPlane)'s default of 64
276 /// MiB; `Some(0)` means no ceiling.
277 ///
278 /// A knob for the same reason `control_plane_workers` is one: the number is
279 /// `memory budget / concurrent inflates`, and the budget is a deployment
280 /// question. It is also the only one of these that turns a request which
281 /// works today into a refusal, which is exactly why it must be reachable
282 /// rather than compiled in.
283 pub max_object_bytes: Option<u64>,
284 /// How often the counter heartbeat is written. `None` turns it
285 /// off.
286 ///
287 /// The heartbeat answers *"what is this server doing right now"* from a log
288 /// the operator already has open, with nothing to scrape. It stays quiet
289 /// while nothing changes, so an idle appliance does not fill its journal.
290 pub metrics_interval: Option<std::time::Duration>,
291 /// Where the **scrape endpoint** binds. `None` does not start it,
292 /// which is the default.
293 ///
294 /// The heartbeat above and this are the same nineteen numbers reached two
295 /// ways, and both are wanted: the heartbeat needs nothing set up and is
296 /// already in the journal when an incident is reconstructed afterwards; the
297 /// endpoint can be *asked*, at whatever rate the person asking chooses,
298 /// which is the only way `transfers_in_flight` distinguishes a slow
299 /// transfer from a stuck one while it is still happening.
300 ///
301 /// Off unless asked for, for the same reason [`ServeConfig::http_bind`] is:
302 /// a second listening port is an operator's decision. See
303 /// [`crate::scrape`] for why it carries no authentication and what
304 /// that does and does not expose.
305 pub metrics_bind: Option<SocketAddr>,
306 /// Replication to a second gunnar (replication). `None` reads
307 /// `<data-root>/replication.json`, which is what an operator configures and
308 /// what a restart picks up; `Some` overrides it in-process and is how a
309 /// test points a primary at a standby it just started on an ephemeral port.
310 ///
311 /// A configuration with no targets starts nothing at all: no task, no
312 /// subscriber. A server that replicates nowhere pays nothing for the
313 /// feature existing.
314 #[cfg(feature = "replication")]
315 pub replication: Option<crate::mirror::ReplicationConfig>,
316 /// **I11's passive heartbeat**: probe this primary and take over if it stops
317 /// answering. `None` — the default — starts no task at all.
318 ///
319 /// Off unless asked for, and that is the design rather than caution:
320 /// `control_plane.rs::replication_abdicate` states the rule, *"naming a new
321 /// primary is the operator's decision, never a default this box picks"*, and
322 /// a box that promoted itself because nobody configured it not to would
323 /// make that decision for every deployment that never heard of the feature.
324 ///
325 /// It does nothing on a box whose role is `Primary`: the task re-reads the
326 /// role every tick and leaves when it is not a standby, so setting it on
327 /// both boxes of a pair is harmless and is what a symmetric unit file wants.
328 /// See [`crate::standby_watch`] for what the probe does and does not prove.
329 #[cfg(feature = "replication")]
330 pub standby_watch: Option<crate::standby_watch::StandbyWatch>,
331 /// **The estate repairs itself** (5.4): the boot check after the control plane
332 /// listens (it RECHARGES only with `auto`, the estate's opt-in, off by default)
333 /// and the pull channel to monetize. `None` (the default): neither runs.
334 /// See [`crate::repair`].
335 #[cfg(feature = "replication")]
336 pub repair: Option<crate::repair::RepairConfig>,
337 /// **Where replication's counters are published to**, for the console band, the
338 /// staleness alarm and `Admin.Stats`.
339 ///
340 /// It is on the config rather than returned from [`Server::start`] because
341 /// the console band starts BEFORE the server does — [`run`] hands the same
342 /// view to both, and `start` drops the replicator into it once there is
343 /// one. Left at its default it reports "nothing configured", which is what
344 /// a server with no standby is.
345 pub replication_view: crate::replication_view::ReplicationView,
346 /// **Where the band's `tenants` row is read from**, the same shape and
347 /// for the same reason as `replication_view`: a metered `Server::start`
348 /// fills it from its entitlement slot; a community build never does, and
349 /// the band draws no row.
350 pub tenants_view: crate::dashboard::TenantView,
351 /// **Where the band's `flow` row is read from.** Filled on every build —
352 /// counting what moved is not the metered appliance's capability. See
353 /// [`crate::dashboard::FlowView`].
354 pub flow_view: crate::dashboard::FlowView,
355 /// What the two git state machines serve under (`gunnar-wire`'s own
356 /// `Config` types, held once — see [`WireConfig`]).
357 ///
358 /// Defaults to `gunnar-wire`'s defaults, which is what every deployment
359 /// silently ran on while `with_config` had no callers. The knobs it carries
360 /// were fully implemented in both protocol versions and reachable from
361 /// nowhere; `--no-filter` in particular is the difference between an
362 /// appliance that declines partial clone and one that only says it might.
363 pub wire: WireConfig,
364 /// **The metered appliance's key and pay page** (`metered` only): what
365 /// `--monetize-pubkey` and `--billing-url` resolve to. `None` on a metered
366 /// build starts it with every namespace Free and `Entitlement.Set` refusing
367 /// by name — never with a key it did not have.
368 #[cfg(feature = "metered")]
369 pub metered: Option<crate::entitlement::MeteredConfig>,
370 /// **The disk-headroom floor** (`--disk-floor`): the community plan gate,
371 /// first link of every front door's chain and the guard on `Repos.Create`.
372 /// See `gunnar_ssh::headroom`. Always present — the default is 5 % or
373 /// 2 GiB, whichever is larger — because a box with no floor is a box that
374 /// finds out it is full from a torn pack.
375 pub disk_floor: gunnar_ssh::DiskFloor,
376 /// **What measures the data set** for the floor and the space ledger.
377 /// `None` — every deployment — is one `statvfs` on the data root. A test
378 /// that must fill a data set of a stated size without a filesystem of that
379 /// size hands its own probe here (`tests/push_space_admission.rs`'s
380 /// virtual disk); nothing else about the server changes.
381 pub space_probe: Option<Arc<dyn gunnar_ssh::SpaceProbe>>,
382}
383
384/// The `--anonymous` policy type, re-exported so a caller that builds a
385/// [`ServeConfig`] (the test harness, `gunnar-cli`) names it from here.
386pub use gunnar_ssh::AnonymousPolicy;
387
388impl Default for ServeConfig {
389 fn default() -> Self {
390 ServeConfig {
391 ssh_bind: SocketAddr::from(([0, 0, 0, 0], gunnar_ssh::config::DEFAULT_PORT)),
392 grpc_bind: BindSpec::Addr(gunnar_grpc::DEFAULT_BIND),
393 grpc_bind_device: None,
394 admin_key: None,
395 restore_pubkeys: Vec::new(),
396 no_grpc: false,
397 wg_only: false,
398 twin_role: false,
399 allow_anonymous: true,
400 anonymous: gunnar_ssh::AnonymousPolicy::Allow,
401 #[cfg(feature = "http")]
402 http_bind: None,
403 #[cfg(feature = "http")]
404 http_cleartext: gunnar_http::Cleartext::default(),
405 #[cfg(feature = "http")]
406 http_tls: None,
407 grpc_tls: None,
408 web_relying_parties: Vec::new(),
409 identity_root_site: false,
410 web_require_user_verification: false,
411 grpc_replication_bind: None,
412 runner: None,
413 max_concurrent_transfers: None,
414 memory_budget_bytes: None,
415 admission_wait: None,
416 max_transfers_per_principal: None,
417 max_anonymous_transfers: None,
418 #[cfg(feature = "admit-share")]
419 admit: gunnar_ssh::edda_admit::Settings::default(),
420 per_principal_limit_is_hard: false,
421 control_plane_workers: None,
422 max_object_bytes: None,
423 metrics_interval: Some(METRICS_INTERVAL),
424 metrics_bind: None,
425 #[cfg(feature = "replication")]
426 replication: None,
427 #[cfg(feature = "replication")]
428 standby_watch: None,
429 #[cfg(feature = "replication")]
430 repair: None,
431 replication_view: crate::replication_view::ReplicationView::new(),
432 tenants_view: crate::dashboard::TenantView::new(),
433 flow_view: crate::dashboard::FlowView::new(),
434 wire: WireConfig::default(),
435 #[cfg(feature = "metered")]
436 metered: None,
437 disk_floor: gunnar_ssh::DiskFloor::default(),
438 space_probe: None,
439 }
440 }
441}
442
443impl ServeConfig {
444 /// A configuration whose listeners take kernel-assigned loopback ports.
445 /// A constructor for tests, not a mode: nothing about what is authenticated
446 /// or dispatched changes.
447 pub fn ephemeral() -> Self {
448 ServeConfig {
449 ssh_bind: SocketAddr::from(([127, 0, 0, 1], 0)),
450 grpc_bind: BindSpec::Addr(SocketAddr::from(([127, 0, 0, 1], 0))),
451 grpc_bind_device: None,
452 ..ServeConfig::default()
453 }
454 }
455}
456
457/// How long [`Server::shutdown`] lets the gRPC listener finish gracefully
458/// before it is aborted.
459///
460/// Long enough for an in-flight unary call to complete on a loaded box, short
461/// enough that a Ctrl-C is answered while the operator is still watching. See
462/// [`Server::shutdown`] for why any bound is needed at all.
463pub const GRPC_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
464
465/// How often `gunnar serve` writes its counters to the log.
466///
467/// Sixty seconds: often enough that an operator watching an incident sees the
468/// numbers move while they are still watching, rare enough that a year of
469/// journal on an idle appliance is not made of heartbeats. The line is skipped
470/// when nothing has changed since the last one, so "quiet" and "not running"
471/// stay distinguishable. A *busy* server is never quiet.
472pub const METRICS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
473
474/// The size of the blocking pool that carries git transfers.
475///
476/// **A decision, not a default.** Every in-flight transfer occupies one blocking
477/// thread for its whole duration (`WireTransport::serve` bridges the async
478/// duplex into `gunnar-wire`'s blocking `BufRead`/`Write` through
479/// `SyncIoBridge`), so this number *was* the server's real concurrency ceiling,
480/// and it was 512 because that is what `tokio` picks when nobody says. Past it,
481/// transfers queued with no log, no counter and no error.
482///
483/// The ceiling that matters is now
484/// [`SshConfig::max_concurrent_transfers`](gunnar_ssh::SshConfig::max_concurrent_transfers),
485/// which refuses visibly. This pool has to stay comfortably **above** it: it
486/// also carries every other blocking thing the server does — pack indexing, ref
487/// transactions, the arrow log — and a pool sized exactly to the transfer limit
488/// would deadlock the moment a transfer needed a second blocking task.
489/// `the_blocking_pool_is_larger_than_the_transfer_limit` asserts the ordering
490/// rather than leaving it to this paragraph.
491pub const GIT_BLOCKING_THREADS: usize = 512;
492
493/// Worker threads that RUN the control plane's calls, derived from the
494/// machine.
495///
496/// A quarter of the cores, at least four and at most eight. The control plane is
497/// an administrative surface — a handful of operators and a UI — not the
498/// serving path, and the whole point of giving it a separate pool is that stalls
499/// stay inside a bounded number of threads. Giving it one core's worth would
500/// make a single slow `StorageInfo` serialise the UI; giving it the machine
501/// would recreate the problem with extra steps. The floor was two until
502/// R5.1-G (2026-10-07): two slow synchronous calls held both, and every other
503/// call — the console's pages, monetize's `Entitlement.Set` — queued behind
504/// them.
505///
506/// These are the HANDLER workers ([`ServiceContext::with_handler_runtime`]).
507/// The HTTP/2 connections run on [`CONTROL_CONNECTION_WORKERS`] workers of
508/// their own, which no handler can hold, so a client's keepalive PING is
509/// answered however long the calls take.
510pub fn control_plane_workers() -> usize {
511 let cores = std::thread::available_parallelism()
512 .map(|n| n.get())
513 .unwrap_or(4);
514 (cores / 4).clamp(4, 8)
515}
516
517/// Worker threads that drive the control plane's HTTP/2 connections (R5.1-G).
518///
519/// They accept, read and write frames — and answer a client's keepalive PING —
520/// and nothing else: every call runs on the handler runtime
521/// ([`control_plane_workers`] workers). They used to be one runtime, and when
522/// the slow synchronous calls held all of its workers nothing answered the
523/// PING; the console and monetize then killed the connection (their keepalive:
524/// a PING every 10 s, dead 5 s later) and every call in flight on it failed.
525pub const CONTROL_CONNECTION_WORKERS: usize = 2;
526
527/// How many CPUs this process may **run on**, from its affinity mask.
528///
529/// `/proc/self/status`'s `Cpus_allowed_list`, because that is the kernel's own
530/// answer and not a request anybody made. `Some(n)` or `None` when the file is
531/// not there to be read; a missing answer is never guessed at, because the whole
532/// point of [`report_cpu_budget`] is a comparison and half a comparison is
533/// worse than none.
534fn cpus_allowed() -> Option<usize> {
535 let status = std::fs::read_to_string("/proc/self/status").ok()?;
536 let list = status
537 .lines()
538 .find_map(|line| line.strip_prefix("Cpus_allowed_list:"))?;
539 count_cpu_list(list)
540}
541
542/// How many CPUs a kernel CPU list names: `0-31`, `3`, `0-3,8,10-11`.
543///
544/// Split out of [`cpus_allowed`] so it can be exercised on lists this host does
545/// not have. It matters: the reference desktop reports a plain `0-31`, so a version of this that
546/// counted ranges and dropped the singletons passed every test on this box
547/// while answering **0 allowed CPUs on a server pinned to one core** — the
548/// condition the caller exists to detect, read exactly backwards.
549fn count_cpu_list(list: &str) -> Option<usize> {
550 let mut total = 0usize;
551 for range in list.trim().split(',') {
552 let range = range.trim();
553 if range.is_empty() {
554 continue;
555 }
556 total += match range.split_once('-') {
557 Some((lo, hi)) => {
558 let (lo, hi) = (
559 lo.trim().parse::<usize>().ok()?,
560 hi.trim().parse::<usize>().ok()?,
561 );
562 hi.checked_sub(lo)?.checked_add(1)?
563 }
564 None => {
565 range.parse::<usize>().ok()?;
566 1
567 }
568 };
569 }
570 (total > 0).then_some(total)
571}
572
573/// Whether the affinity is wide enough against the quota to be worth a line,
574/// and the two numbers to say it with.
575///
576/// Pure, and given both readings rather than taking them, so the condition can
577/// be seen in both states from a test on any machine. Either reading missing is
578/// `None`: half a comparison is worse than none.
579///
580/// **Twice, not any excess.** The two numbers are rarely exactly equal and a
581/// line that fires on nine-versus-eight is a line operators learn to skip.
582/// Twice is where the measurement above already showed a real cost.
583fn spread_budget(allowed: Option<usize>, quota: Option<f64>) -> Option<(usize, usize)> {
584 let (allowed, quota) = (allowed?, quota?);
585 let budget = quota.ceil().max(1.0) as usize;
586 (allowed >= budget.saturating_mul(2)).then_some((allowed, budget))
587}
588
589/// How many CPUs' worth of time this process may **use**, from cgroup v2
590/// `cpu.max`. `None` when there is no quota, which is the ordinary case.
591fn cpu_quota() -> Option<f64> {
592 let own = std::fs::read_to_string("/proc/self/cgroup").ok()?;
593 let rel = own
594 .lines()
595 .find_map(|line| line.strip_prefix("0::"))?
596 .trim()
597 .trim_start_matches('/');
598 let path = std::path::Path::new("/sys/fs/cgroup")
599 .join(rel)
600 .join("cpu.max");
601 let raw = std::fs::read_to_string(path).ok()?;
602 let mut parts = raw.split_whitespace();
603 let quota = parts.next()?;
604 if quota == "max" {
605 return None;
606 }
607 let quota: f64 = quota.parse().ok()?;
608 let period: f64 = parts.next()?.parse().ok()?;
609 (period > 0.0).then(|| quota / period)
610}
611
612/// **Say when the box will let this server run far wider than it will let it
613/// work**, because that costs more than anything in the request path.
614///
615/// A container started with `--cpus 8` on a 32-core host gets eight CPUs of
616/// *quota* spread over thirty-two CPUs of *affinity*. gunnar serves one transfer
617/// on three threads that hand work to each other — a hyper task on a runtime
618/// worker, the `spawn_blocking` thread that runs the git state machine, and the
619/// pack-entry worker `iter_from_counts` spawns — so at N concurrent transfers
620/// the scheduler has ~3N runnable threads and thirty-two places to put them,
621/// while only eight of them may run at a time. Every hand-off then crosses a
622/// core, and a hand-off that crosses a core is a futex.
623///
624/// Measured on the reference desktop, 2026-08-04, one binary, one 8-CPU budget, the `year`
625/// corpus (28 893 objects) over smart HTTP, ramp 1→32 clients:
626///
627/// | budget applied as | CPU-seconds per request, 1 client → 32 |
628/// |---|---|
629/// | `cpu.max` quota only, 32 CPUs allowed | 0.42 → 0.94 |
630/// | the same quota, affinity pinned to 8 | 0.34 → 0.35 |
631///
632/// The quota-only arm spent **40% of the server's CPU** in the kernel's futex
633/// hash-bucket spinlock (`native_queued_spin_lock_slowpath`, `perf record` on
634/// the serving process). The pinned arm is flat: gunnar's per-request cost does
635/// not in fact grow with concurrency, and the ramp that said it did was
636/// measuring where the threads were allowed to run.
637///
638/// So this is a line in the log and not a change of behaviour: gunnar must not
639/// pin itself. It does not know which CPUs are *its* to take, and choosing wrong
640/// would be worse than the condition being reported. The operator does know, and
641/// `--cpuset-cpus` is the whole fix.
642///
643/// The decision is [`spread_budget`], separately, because a condition read off
644/// the host it is running on can only ever be seen in the state that host
645/// happens to be in — and this one has to be seen in both.
646fn report_cpu_budget() {
647 let Some((allowed, budget)) = spread_budget(cpus_allowed(), cpu_quota()) else {
648 return;
649 };
650 tracing::warn!(
651 cpus_allowed = allowed,
652 cpu_quota = budget,
653 "this server may run on {allowed} CPUs but may only use {budget} CPUs' worth of \
654 them. Every hand-off between the three threads that serve one transfer then \
655 crosses a core, and in a 2026-08-04 measurement that cost 2.7x the CPU per \
656 request at 32 concurrent clients against the identical budget applied as a \
657 cpuset. Start the \
658 container with `--cpuset-cpus` (or the unit with `AllowedCPUs=`) naming {budget} \
659 CPUs, not only `--cpus`/`CPUQuota=`."
660 );
661}
662
663/// The runtime `gunnar serve` runs the **git** listener on.
664///
665/// Built here rather than with `#[tokio::main]` so a client-only build links no
666/// async runtime at all, and built with an explicit
667/// [`GIT_BLOCKING_THREADS`] so the number that used to be the server's
668/// accidental concurrency ceiling is now written down somewhere.
669pub fn runtime() -> Result<tokio::runtime::Runtime> {
670 tokio::runtime::Builder::new_multi_thread()
671 .max_blocking_threads(GIT_BLOCKING_THREADS)
672 .thread_name("gunnar-git")
673 .enable_all()
674 .build()
675 .context("building the git runtime")
676}
677
678/// Both listeners, bound and running.
679/// How the smart-HTTP listener gets its TLS identity (see
680/// [`ServeConfig::http_tls`]).
681#[cfg(feature = "http")]
682#[derive(Debug, Clone)]
683pub enum HttpTls {
684 /// Mint a leaf from this server's own issuer at start — the
685 /// single-executable deployment. Refuses to start if it cannot mint;
686 /// silent cleartext fallback is the defect this enum exists to prevent.
687 SelfMinted,
688 /// Operator-supplied chain and key.
689 Identity(gunnar_http::TlsIdentity),
690}
691
692/// How the gRPC control plane gets its **mutual** TLS material.
693///
694/// The mirror of [`HttpTls`], with one difference that is the whole point: the
695/// smart-HTTP listener terminates TLS to encrypt, and this one terminates TLS
696/// to *identify*. So there is no bring-your-own-CA arm here yet. A certificate
697/// the gate can turn into a principal has to have been registered in the ACL
698/// table, and the only thing that mints and registers in one act is this
699/// server's own issuer (`Bootstrap.Enroll`, `gunnar control bootstrap`).
700/// Trusting a foreign CA would mean a certificate that passes the handshake and
701/// can never pass the gate, which is a confusing way to be refused.
702#[derive(Debug, Clone, Copy, PartialEq, Eq)]
703pub enum GrpcTls {
704 /// Mint the listener's own leaf from this server's issuer at start, and
705 /// pin that issuer as the CA every client certificate must come from.
706 ///
707 /// Refuses to start if it cannot mint. A control plane that asked for
708 /// client certificates and came up without them would be the same silent
709 /// downgrade [`HttpTls`] refuses.
710 SelfMinted,
711}
712
713/// **A tokio `Runtime` that is always torn down with `shutdown_background`,
714/// never by a blocking `Drop`.**
715///
716/// Dropping a `Runtime` blocks the current thread on its worker and blocking
717/// pools' shutdown, and tokio **panics** when that blocking drop happens inside
718/// another runtime's context: *"Cannot drop a runtime in a context where
719/// blocking is not allowed."*
720///
721/// [`Server::start`] is an `async fn` that builds the control plane's own
722/// runtime and then runs several more fallible steps before it moves that
723/// runtime into [`Server`]. A `?` anywhere in that window used to drop a bare
724/// `Runtime` on the async stack — and on the appliance gunnar is **PID 1**, so
725/// that panic is the kernel's `Attempted to kill init`. Measured 2026-08-31:
726/// a power cut left a 0-byte `replication.json`, its parse in `start`
727/// failed, the `?` unwound past the live runtime, and the box wedged on every
728/// reboot. Wrapping the runtime in this guard makes the early return safe **by
729/// construction** — one drop path, and it never blocks — instead of by
730/// remembering to hand-`take()` it at each of the several exits.
731///
732/// Since R5.1-G it holds the control plane's TWO runtimes — the connections'
733/// and the calls' — under the one guard, in that order.
734#[derive(Default)]
735struct ControlRuntime(Vec<tokio::runtime::Runtime>);
736
737impl ControlRuntime {
738 /// Hold a runtime whose drop must be non-blocking.
739 fn new(runtime: tokio::runtime::Runtime) -> Self {
740 ControlRuntime(vec![runtime])
741 }
742
743 /// Take over every runtime `other` holds, so one guard drops them all.
744 fn hold(&mut self, other: &mut ControlRuntime) {
745 self.0.append(&mut other.0);
746 }
747
748 /// A handle on the first runtime held, if any.
749 fn handle(&self) -> Option<tokio::runtime::Handle> {
750 self.0.first().map(|runtime| runtime.handle().clone())
751 }
752
753 /// Tear the runtimes down now, without blocking, if any are still held.
754 /// Idempotent, so an explicit call and the `Drop` cannot double-free.
755 fn shutdown(&mut self) {
756 for runtime in self.0.drain(..) {
757 runtime.shutdown_background();
758 }
759 }
760}
761
762impl Drop for ControlRuntime {
763 fn drop(&mut self) {
764 self.shutdown();
765 }
766}
767
768pub struct Server {
769 /// An `Option` only because [`Server`] now has a [`Drop`] impl, which makes
770 /// moving a field out of `self` in [`Server::shutdown`] illegal.
771 ssh: Option<BoundServer>,
772 grpc: Option<JoinHandle<Result<()>>>,
773 grpc_shutdown: Option<oneshot::Sender<()>>,
774 /// The bound gRPC address and whether that listener authenticates, as
775 /// ONE field. They are printed together and must not be able to disagree.
776 grpc_endpoint: Option<(SocketAddr, bool)>,
777 /// The control plane's **own** runtime, kept alive here because
778 /// dropping it would stop the listener it carries.
779 ///
780 /// A [`ControlRuntime`] guard, not a bare `Runtime`: dropping a `Runtime`
781 /// blocks, and doing so from inside another runtime's context panics
782 /// ("Cannot drop a runtime in a context where blocking is not allowed"),
783 /// and both [`Server::start`] and [`Server::shutdown`] run in async
784 /// context. The guard routes *every* drop through `shutdown_background`, so
785 /// no `?`, panic or forgotten `take()` can turn a teardown into an abort —
786 /// which as PID 1 is a kernel panic.
787 control_runtime: ControlRuntime,
788 stores: Arc<ServerStores>,
789 /// Whether the unauthenticated `Bootstrap.Enroll` window was open when this
790 /// server started, so the banner on stdout says the same thing the log does.
791 ///
792 /// A field rather than a re-read: the banner and the warning must not be
793 /// able to disagree about the single most important fact a fresh server has
794 /// (one implementation).
795 bootstrap_window_open: bool,
796 /// **The SSH host key this server is serving, `SHA256:…`.**
797 ///
798 /// Kept so the startup banner can print it. It is logged at start through
799 /// `tracing`, which on the appliance means stderr, which means
800 /// `/dev/console`, which is the LAST `console=` on the command line, which
801 /// is the serial line — and the UpCloud shape has no serial line. So the
802 /// one emission this box makes of the one fact a client has to pin
803 /// reached nobody in production, and `known_hosts` got filled by scanning
804 /// instead. A field, not a re-read, for the same reason
805 /// `bootstrap_window_open` is one: the banner and the log must not be able
806 /// to disagree about it.
807 host_key_fingerprint: String,
808 /// The counters both listeners write to.
809 metrics: Arc<ServerMetrics>,
810 /// The heartbeat task, aborted at shutdown so a stopped server stops
811 /// narrating.
812 heartbeat: Option<JoinHandle<()>>,
813 /// I11's standby watch, held so it is aborted with the server. A watchdog
814 /// that outlived the box it watches over could promote a server that is
815 /// shutting down.
816 #[cfg(feature = "replication")]
817 standby_watch: Option<JoinHandle<()>>,
818 /// The boot check and the pull channel ([`crate::repair`]), aborted with the
819 /// server (a recharge it was running fails its restore stream, never ends it).
820 #[cfg(feature = "replication")]
821 repair: Option<JoinHandle<()>>,
822 /// The event bus both the transport and the control plane publish to.
823 bus: Arc<EventBus>,
824 /// The smart-HTTP listener's task and the address it bound (smart HTTP).
825 /// Dropped with the `Server`, and therefore aborted.
826 #[cfg(feature = "http")]
827 http: Option<(JoinHandle<std::io::Result<()>>, SocketAddr, bool)>,
828 /// The scrape listener's task and the address it bound. Dropped
829 /// with the `Server`, and therefore aborted: a stopped server must stop
830 /// answering, or a scraper keeps drawing the last numbers it saw.
831 scrape: Option<(JoinHandle<std::io::Result<()>>, SocketAddr)>,
832 /// The replicator, when any standby is configured.
833 /// the replicator and its workers, in the LIVE SLOT the control plane
834 /// reconfigures through.
835 ///
836 /// It was an `Option<Arc<Replicator>>` plus an `Option<ReplicationTasks>`
837 /// until 2026-08-30 — decided once at startup, from a file. That is
838 /// unusable on the shape this ships to: an OS-less appliance with no shell,
839 /// where the only way to name a standby is over the control plane. See
840 /// [`ReplicationSlot`](crate::mirror::ReplicationSlot).
841 ///
842 /// Dropped with the `Server`, and therefore aborted, so a stopped server
843 /// stops replicating.
844 #[cfg(feature = "replication")]
845 replication: Arc<crate::mirror::ReplicationSlot>,
846 /// The outbound webhook delivery slot, reconfigured through the control plane.
847 /// Dropped with the `Server`, and therefore aborted, so a stopped server
848 /// stops delivering.
849 #[cfg(feature = "webhooks")]
850 webhook: Arc<crate::webhook::WebhookSlot>,
851}
852
853/// One line every `interval` naming what the server has done, but only when
854/// something moved.
855///
856/// **Not a scrape endpoint.** The load target is a forge, and the first
857/// question at an incident gets answered from the log the operator already has
858/// open. A Prometheus surface is a separate decision with its own cost: another
859/// listener, and another thing to authenticate. This makes the counters
860/// readable today, in four lines.
861async fn heartbeat(metrics: Arc<ServerMetrics>, interval: std::time::Duration) {
862 let mut ticker = tokio::time::interval(interval);
863 // The first tick fires immediately; skip it so start-up does not emit a row
864 // of zeroes that reads like traffic.
865 ticker.tick().await;
866 let mut previous = metrics.snapshot();
867 loop {
868 ticker.tick().await;
869 let now = metrics.snapshot();
870 // An idle server says nothing. A server that has served ANYTHING since
871 // the last beat says what, including one whose only change is that a
872 // transfer finished; that is how `transfers_in_flight` coming back to
873 // zero becomes visible.
874 if now == previous {
875 continue;
876 }
877 emit_metrics(&now);
878 previous = now;
879 }
880}
881
882/// The gRPC control plane's request ids, minted from the **same** counter the
883/// SSH listener and the smart-HTTP listener mint theirs from.
884///
885/// This one line is why `gunnar_grpc::router` takes a source instead of owning
886/// one: `gunnar-grpc` has no `gunnar-core` edge and must not grow one (it would
887/// put `aws-lc-rs`, C and a build script into the `--features client,control`
888/// graph), and a private counter over there would mean a control-plane call and
889/// a git transfer could print the same `req=` value in one log.
890///
891/// `gunnar_grpc::trace`'s rendering is asserted equal to
892/// [`RequestId`]'s by `the_grpc_request_id_renders_exactly_as_gunnar_core_
893/// renders_one` in `tests/observability.rs`, because a shared counter rendered
894/// two ways is still two identifiers.
895fn grpc_request_id() -> u64 {
896 RequestId::next().get()
897}
898
899/// The heartbeat line.
900///
901/// **This function used to BE the line**, and it listed eighteen of the
902/// nineteen counters. `connections_anonymous` — written by both transports
903/// (`gunnar-ssh/src/server.rs`, `gunnar-http/src/service.rs`), and the answer to
904/// *how much of my traffic is strangers*, which the field's own documentation
905/// calls the first question an operator asks after making anything public —
906/// appeared in no line gunnar had ever printed. Nothing was wrong with the
907/// counter; the hand-written list next to it had simply never been updated when
908/// the field was added.
909///
910/// So it is not a hand-written list any more. The line expands from the same
911/// macro arm that declares the fields, in `gunnar-core`, beside the scrape
912/// endpoint's renderer, and neither can fall behind the struct (one implementation). This
913/// wrapper stays only so the heartbeat task and a test call the same one thing.
914fn emit_metrics(now: &MetricsSnapshot) {
915 now.emit_heartbeat();
916}
917
918impl Server {
919 /// Bind everything and start accepting.
920 ///
921 /// Returns once both sockets are bound, so a caller knows the real
922 /// addresses before the first client connects, which is what makes port 0
923 /// usable in a test.
924 pub async fn start(stores: Arc<ServerStores>, config: ServeConfig) -> Result<Server> {
925 report_cpu_budget();
926 // The bind spelling becomes an address HERE, before anything else in
927 // this function can report one. `run` has already resolved it and
928 // replaced the spec with the answer, so on the appliance path this is
929 // the identity case; a caller that reaches `start` directly (every
930 // test, and `gunnar serve` under an init system) gets the resolution
931 // it would otherwise have had to do itself.
932 let mut config = config;
933 // A bind by ROLE (`utility:50051`, the lean estate) becomes this box's NIC first.
934 config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)?;
935 // The interface spelling names a DEVICE as well as an address; keep the
936 // device before the spec becomes an address, so the listener can bind it.
937 if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
938 config.grpc_bind_device = Some(interface.clone());
939 }
940 // **An interface whose lease PID 1 still asks for** (its boot DHCP failed —
941 // estate-box3 D4, a box rebooted while its network was down) is WAITED for: the
942 // control plane binds once the address arrives instead of never.
943 if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
944 let has_address =
945 |i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
946 if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
947 tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
948 while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
949 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
950 }
951 }
952 }
953 // ★ **An estate box is WireGuard-only** (owner 2026-10-04: every
954 // box-to-box connection over the tunnel), whatever its medium says: a
955 // pairing seed with a WireGuard tunnel IS an estate (the front, the
956 // relay, the overlay). A standalone appliance — no seed — is not.
957 #[cfg(feature = "wg-client")]
958 if !config.wg_only {
959 if let Ok(Some(seed)) = crate::pairing::seed(stores.root()) {
960 if crate::tunnel_addr::estate_is_wg_only(Some(
961 seed.tunnel == nordisk_pairing::seed::TunnelKind::Wireguard,
962 )) {
963 config.wg_only = true;
964 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");
965 }
966 }
967 }
968 // The posture the pairing reads (which address of the primary a twin joins).
969 crate::tunnel_addr::set_wg_only(config.wg_only);
970 let resolved_grpc_bind = match resolve_grpc_bind(&config.grpc_bind)? {
971 Some(addr) => addr,
972 None => {
973 // The interface is there and holds no address. There is nothing
974 // to bind, so there is no control plane — the same shape
975 // `--no-grpc` produces, reached for the same honest reason and
976 // reported through the same field, so the dashboard row, the
977 // readiness line and `Admin.Health` all agree.
978 config.no_grpc = true;
979 gunnar_grpc::DEFAULT_BIND
980 }
981 };
982 // ── the bootstrap: refuse to serve rather than start unreachable ───
983 let admin_key = match &config.admin_key {
984 Some(spec) => Some(read_key_argument(spec)?),
985 None => None,
986 };
987 // The keys a restore is taken from, resolved and CHECKED now: a line that does not
988 // parse would otherwise be a key that silently lets nobody in, found out on the day
989 // of the restore.
990 let restore_keys = config
991 .restore_pubkeys
992 .iter()
993 .map(|spec| {
994 let line = read_key_argument(spec)?;
995 gunnar_core::auth::parse_public_key(line.trim()).map_err(|err| {
996 anyhow::anyhow!(
997 "--restore-pubkey {spec:?} is not an OpenSSH public key line: {err:#}"
998 )
999 })?;
1000 Ok(line.trim().to_owned())
1001 })
1002 .collect::<Result<Vec<String>>>()?;
1003 let readiness = stores
1004 .auth()
1005 .prepare_to_serve(admin_key.as_deref())
1006 .with_context(|| {
1007 format!(
1008 "reading the identity table at {}. This is NOT the same as finding it \
1009 empty: a data path that cannot be read must never be served as a server \
1010 nobody has enrolled yet, because that would reopen the bootstrap window on \
1011 a disk fault.",
1012 stores.root().display()
1013 )
1014 })?;
1015 announce_readiness(
1016 &readiness,
1017 stores.root(),
1018 resolved_grpc_bind,
1019 config.no_grpc,
1020 );
1021 let bootstrap_window_open = readiness.bootstrap_window_open();
1022
1023 // ── the SSH host key, minted in-process and persisted ──────────────
1024 // A twin frames nothing: its host key is read (the primary's, replayed)
1025 // or minted through the UNLOGGED store, and its follower starts here.
1026 let host_key_store = if stores.is_twin() {
1027 stores.secrets_unframed()?
1028 } else {
1029 stores.secrets()?
1030 };
1031 let host_key =
1032 ensure_host_key(&host_key_store).context("loading or minting the SSH host key")?;
1033 if stores.is_twin() {
1034 crate::twin::start_follower(Arc::clone(&stores));
1035 }
1036 // The pairing loop, once the control plane serves: join the front,
1037 // take the role, become the twin if elected (pairing.rs).
1038 #[cfg(feature = "wg-client")]
1039 match crate::pairing::seed(stores.root()) {
1040 Ok(Some(seed)) => crate::pairing::start(Arc::clone(&stores), seed),
1041 Ok(None) => {}
1042 Err(e) => {
1043 tracing::error!(error = %format!("{e:#}"), "pairing: the seed is present and unusable — this box will not pair")
1044 }
1045 }
1046 let host_key_fingerprint = host_key.fingerprint().to_string();
1047 tracing::info!(
1048 fingerprint = %host_key_fingerprint,
1049 "ssh host key ready (persisted; it survives a restart)"
1050 );
1051 // The fingerprint crosses the ssh-key version split as BYTES: gunnar-core
1052 // links ssh-key 0.6 and russh pins 0.7.0-rc.10, so no key type can be
1053 // handed from one to the other.
1054 let russh_key = russh_private_key(host_key.private_openssh())
1055 .context("handing the host key to russh")?;
1056
1057 // ── the certificate issuer, minted at first start ──────────────────
1058 //
1059 // The same discipline as the host key above: a fresh appliance mints
1060 // its own, and a restart finds the one it minted. `Pki::create`'s
1061 // refusal lives under the document's compare-and-swap, so two racing
1062 // starts cannot mint two issuers. Failure is a warning, not a refusal
1063 // to serve: SSH-key auth carries the server whether or not certificates
1064 // exist, and an operator who imports a CA later gets exactly the
1065 // `issuer import` refusal they should.
1066 match stores.pki() {
1067 Ok(pki) => match pki.issuer() {
1068 Ok(Some(live)) => tracing::info!(
1069 subject = %live.subject,
1070 fingerprint = %live.fingerprint,
1071 "issuer ready (persisted; it survives a restart)"
1072 ),
1073 Ok(None) => match pki.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS) {
1074 Ok(created) => tracing::warn!(
1075 subject = %created.subject,
1076 fingerprint = %created.fingerprint,
1077 "FIRST START: this server minted its own certificate issuer. Its \
1078 private key is sealed in the secret store. `gunnar issuer show --pem` \
1079 prints the certificate clients pin."
1080 ),
1081 Err(err) => tracing::warn!(
1082 %err,
1083 "the issuer could not be minted at first start; certificate issuance \
1084 is unavailable until `gunnar issuer create` or `issuer import` succeeds"
1085 ),
1086 },
1087 Err(err) => tracing::warn!(%err, "the PKI document could not be read"),
1088 },
1089 Err(err) => tracing::warn!(
1090 %err,
1091 "no secret store, so no issuer: certificate issuance is unavailable"
1092 ),
1093 }
1094
1095 // ── can this server be moved onto mutual TLS? ──────────────────────
1096 //
1097 // Asked HERE, and the position is two decisions rather than one.
1098 //
1099 // **After the issuer block above**, because that block MINTS one on a
1100 // first start: asking before it would refuse a server whose issuer is
1101 // four lines from existing, and the refusal would name the one thing
1102 // the operator did not have to do.
1103 //
1104 // **Before anything binds, and before the control plane's runtime is
1105 // built.** Two reasons:
1106 //
1107 // 1. `--grpc-tls-self` is a one-way door for everybody who cannot
1108 // complete the handshake afterwards, and exactly one state makes
1109 // it a door with nobody on either side. A server asked for client
1110 // certificates that cannot honour the ask must be a startup error
1111 // naming the cause, never a listener that came up — the same
1112 // discipline `--http-tls-*` already follows.
1113 // 2. **A `?` past the control runtime is an appliance that will not
1114 // boot.** `ControlRuntime`'s own docs record the measurement: that
1115 // `Runtime` is built bare inside this function and dropping one on
1116 // an async stack panics, and on the appliance gunnar is PID 1, so
1117 // the panic is `Attempted to kill init`. The first draft of this
1118 // check sat beside the trust root, inside that window, and
1119 // `tests/tls_transition.rs` failed with exactly that panic instead
1120 // of the refusal it was asserting. That test is the guard that it
1121 // does not move back.
1122 //
1123 // `crate::tls_preflight` holds the argument and the verdicts;
1124 // `gunnar issuer readiness` is this same function asked in advance,
1125 // so the answer an operator plans against is the answer acted on here
1126 // (one implementation).
1127 if !config.no_grpc && config.grpc_tls.is_some() {
1128 let preflight = crate::tls_preflight::preflight(&stores)
1129 .context("checking whether this server can move to mutual TLS")?;
1130 if let Some(refusal) = preflight.refusal() {
1131 anyhow::bail!("{refusal}");
1132 }
1133 preflight.warn();
1134 }
1135
1136 // ── the SSH listener ───────────────────────────────────────────────
1137 //
1138 // ONE set of counters, minted here and shared by the listener
1139 // and the transport. Two `ServerMetrics` would produce two partial
1140 // truths, the listener's refusals in one and the transport's pushes in
1141 // the other, and neither answers "what is this server doing".
1142 // The stores' set (A28 M17): the flow flush's reconcile pass, started before this,
1143 // counts its misses on it too.
1144 let metrics = Arc::clone(stores.metrics());
1145 let bus = Arc::new(EventBus::new(gunnar_grpc::DEFAULT_EVENT_BUFFER));
1146 let transport = Arc::new(
1147 WireTransport::new(Arc::clone(&stores))
1148 .with_bus(Arc::clone(&bus))
1149 .with_metrics(Arc::clone(&metrics))
1150 .with_wire_config(config.wire.clone()),
1151 );
1152 let defaults = SshConfig::default();
1153 // The DECLARED budget (`crate::budget`): cores x TRANSFERS_PER_CORE,
1154 // capped by the descriptors the handle cache has not already spoken
1155 // for, with `MemAvailable` sizing the ledger. An operator's
1156 // `--max-concurrent-transfers` still wins — `.or()` below — and so does
1157 // `GUNNAR_MAX_TRANSFERS`, which is how the appliance's kernel command
1158 // line reaches it. Before this the ceiling was the shipped 256, which
1159 // the governor then walked down to 19 under load while the
1160 // ledger's flat 320 MiB-per-push refused everything past 25.
1161 let mut budgets = crate::budget::budgets();
1162 // **Recorded, so the box can say later whether its socket is where a
1163 // twin's belongs** (2026-09-21, lane T4). A box told it is a twin AFTER
1164 // this line keeps whatever was bound here until it restarts;
1165 // `twin::git_listener_needs_restart` is how that is reported instead of
1166 // being left to `ss`.
1167 let twin_shaped = stores.is_twin() || config.twin_role;
1168 crate::twin::note_git_bind_is_twin_shaped(twin_shaped);
1169 let ssh_config = SshConfig {
1170 bind: if twin_shaped {
1171 // **No forge listener on a twin.** It serves no git to anyone: a
1172 // read would grow the caches the twin must not hold. The listener
1173 // the rest of `start` is wired around stays, on loopback, port 0.
1174 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");
1175 SocketAddr::from(([127, 0, 0, 1], 0))
1176 } else {
1177 config.ssh_bind
1178 },
1179 allow_anonymous: config.allow_anonymous,
1180 // The banner names the BUILD, not just the crate version. On the
1181 // OS-less appliance this listener is the only thing that can be
1182 // asked what is running — there is no shell to run `gunnar
1183 // --version` in. See `gunnar_common::ssh_server_id`.
1184 server_id: gunnar_common::ssh_server_id(),
1185 // `or` rather than `unwrap_or`: `None` here means "the operator did
1186 // not say", which takes the default. Turning a limit *off* is
1187 // `Some(0)`-free and stays a code path, not a flag, because an
1188 // appliance with no transfer ceiling is the unbounded listener
1189 // exactly as it was.
1190 max_concurrent_transfers: config.max_concurrent_transfers.or(Some(budgets.transfers)),
1191 memory_budget_bytes: config.memory_budget_bytes.or(Some(budgets.memory_bytes)),
1192 admission_wait: config.admission_wait.or(defaults.admission_wait),
1193 max_transfers_per_principal: config
1194 .max_transfers_per_principal
1195 .or(defaults.max_transfers_per_principal),
1196 // `or` for the same reason as the two above: `None` means the
1197 // operator did not say, which takes `SshConfig::default`'s
1198 // ceiling. Switching it OFF is deliberately not reachable from a
1199 // flag — `Some(0)` refuses everything anonymous, and letting
1200 // anonymous inherit a keyed identity's share is the hole anonymous read opened
1201 // and this closed, not a configuration.
1202 max_anonymous_transfers: config
1203 .max_anonymous_transfers
1204 .or(defaults.max_anonymous_transfers),
1205 per_principal_limit_is_hard: config.per_principal_limit_is_hard,
1206 #[cfg(feature = "admit-share")]
1207 admit: config.admit,
1208 ..defaults
1209 };
1210 // The row says what is IN FORCE, after the operator's flags: a line
1211 // that printed the derivation while the server ran on something else
1212 // would be worse than no line, because the first thing anybody does
1213 // with it is compare it to the refusals underneath it.
1214 if let Some(n) = config.max_concurrent_transfers {
1215 budgets.transfers = n;
1216 budgets.transfers_from = crate::budget::Source::Operator;
1217 }
1218 if let Some(bytes) = config.memory_budget_bytes {
1219 budgets.memory_bytes = bytes;
1220 }
1221 if let Some(wait) = config.admission_wait {
1222 budgets.wait_ms = wait.as_millis() as u64;
1223 }
1224 crate::budget::announce(&budgets);
1225
1226 #[cfg(feature = "http")]
1227 let http_transport: Arc<dyn gunnar_ssh::seam::GitTransport> = Arc::clone(&transport) as _;
1228 // The SAME `WireTransport`, reached through a second trait rather than
1229 // a second object: `open_lfs` hands out the store under the repository's
1230 // own git dir and the `WireConfig::lfs` limits the SSH session is built
1231 // with, so an operator's `--max-lfs-object-bytes` binds on both
1232 // transports or on neither.
1233 // ── the metered appliance's entitlement slot, before any front door ─
1234 //
1235 // Opened here so BOTH front doors and the control plane share ONE
1236 // slot (one implementation): the same arc-swap the SSH dispatch reads is the one
1237 // `Entitlement.Set` swaps. Loading it re-verifies every stored fact
1238 // (see `crate::entitlement`), and a key that does not parse is a
1239 // startup error — a metered box that silently ran Free would be the
1240 // provider's revenue leaking with no line to say so.
1241 #[cfg(feature = "metered")]
1242 let entitlement = match &config.metered {
1243 Some(metered) => Some(
1244 crate::entitlement::EntitlementSlot::open(Arc::clone(&stores), metered)
1245 .context("opening the entitlement slot")?,
1246 ),
1247 None => {
1248 tracing::warn!(
1249 "metered build started without --monetize-pubkey: every namespace is Free \
1250 and Entitlement.Set refuses every fact"
1251 );
1252 None
1253 }
1254 };
1255 // The band's tenants row reads the SAME slot `Admin.Stats` counts from
1256 // (one implementation). Filled here, before any door and outside the gRPC branch, so
1257 // a `--no-grpc` metered box still shows its tenants.
1258 // **This estate SELLS room** (the feed's `sells_room`, LIVE; gunnar-grpc `selling`):
1259 // ONE flag, read by the entitlement slot's push gate and the gRPC context alike.
1260 #[cfg(feature = "metered")]
1261 let sells_room = entitlement
1262 .as_ref()
1263 .map(|slot| slot.sells_room_flag())
1264 .unwrap_or_else(|| Arc::new(std::sync::atomic::AtomicBool::new(false)));
1265 #[cfg(not(feature = "metered"))]
1266 let sells_room = Arc::new(std::sync::atomic::AtomicBool::new(false));
1267 // **The front's egress door** (the lean estate, A19): a box laid with no public IPv4 is
1268 // fed it, and every client of this box that reaches the internet takes it from then on.
1269 nordisk_pairing::feedbus::on_published("the front's egress door", crate::feed::take_egress);
1270 {
1271 let flag = Arc::clone(&sells_room);
1272 nordisk_pairing::feedbus::on_published(
1273 "shop room: a person's account with no fact",
1274 move |feed| {
1275 let sells_room = crate::feed::Fed::from_feed(feed).sells_room;
1276 if flag.swap(sells_room, std::sync::atomic::Ordering::Relaxed) != sells_room {
1277 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" });
1278 }
1279 },
1280 );
1281 }
1282 #[cfg(feature = "metered")]
1283 if let Some(slot) = &entitlement {
1284 // A key the installer feeds later is applied to THIS slot in place
1285 // (owner 2026-09-25: avoid restarts; `feed::monetize_arrived`).
1286 let taker = Arc::clone(slot);
1287 crate::feed::on_monetize_key(
1288 "entitlement slot",
1289 Box::new(move |k| taker.set_key(k).map(|_| ())),
1290 );
1291 // ★ UW-6: the SITE's pay page, fed per estate (`refusal_link`), replaces the
1292 // constant medium's — now and on every feed after (LIVE).
1293 let pay = Arc::clone(slot);
1294 nordisk_pairing::feedbus::on_published(
1295 "entitlement: the site's pay page",
1296 move |feed| {
1297 if let Some(url) = crate::feed::Fed::from_feed(feed).billing_url.as_deref() {
1298 if let Err(why) = pay.set_billing_url(url) {
1299 tracing::warn!(target: "gunnar::audit", event = "entitlement.billing_url_refused", reason = %why, "the fed pay page was refused; the one in force stays");
1300 }
1301 }
1302 },
1303 );
1304 let slot = Arc::clone(slot);
1305 config.tenants_view.fill(move || slot.counts());
1306 }
1307 // ── spools a crash left behind (cov-13) ──
1308 //
1309 // A pushed pack of 64 MiB or more is spooled to `incoming-*.pack` while
1310 // it arrives and removed by `Drop` when it lands or fails. A process
1311 // that died mid-receive ran no `Drop`, so the file is still here, full
1312 // size, under a name nothing reads — room the next push is refused for.
1313 // No door is open yet, so no live push owns one.
1314 let (mut swept, mut swept_bytes) =
1315 crate::push_space::sweep_spool_leftovers(stores.repos_root());
1316 // …and an LFS upload's `<txlog>/spool/*.part`, the same shape one door over.
1317 let (lfs, lfs_bytes) = crate::push_space::sweep_lfs_spool(
1318 &stores.txlog().dir().join(gunnar_core::txlog::SPOOL_DIR),
1319 );
1320 swept += lfs;
1321 swept_bytes += lfs_bytes;
1322 if swept > 0 {
1323 tracing::warn!(
1324 files = swept,
1325 bytes = swept_bytes,
1326 "deleted {swept} pushed-pack spool file(s) a crash left behind ({})",
1327 gunnar_ssh::headroom::human(swept_bytes)
1328 );
1329 }
1330 // ── the disk-headroom gate: the community plan gate, every build ──
1331 //
1332 // Measured once here so the boot log states the floor IN FORCE beside
1333 // what is free right now; refused per request in the chain below and
1334 // in `Repos.Create`. A root that cannot be measured is logged as the
1335 // refusal it will be, not hidden until the first push finds it.
1336 //
1337 // ★ The filesystem's preallocation, measured once here (round 5.2): what a growing
1338 // spool-sized file occupies beyond its bytes on THIS data set (XFS doubles it; tmpfs and
1339 // ext4 add nothing), so every push reserves what it will really hold while it lands.
1340 let (fs_slack, unmeasured) = gunnar_ssh::space::FsSlack::of(stores.root());
1341 match &unmeasured {
1342 None => {
1343 tracing::info!(root = %stores.root().display(), measured_permille = ?fs_slack.measured_permille, slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), None))
1344 }
1345 Some(why) => {
1346 tracing::warn!(root = %stores.root().display(), slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), Some(why)))
1347 }
1348 }
1349 let ledger = match &config.space_probe {
1350 None => gunnar_ssh::SpaceLedger::new(stores.root(), config.disk_floor),
1351 Some(probe) => gunnar_ssh::SpaceLedger::with_probe(
1352 stores.root(),
1353 config.disk_floor,
1354 Arc::clone(probe),
1355 ),
1356 };
1357 let disk_gate = Arc::new(gunnar_ssh::DiskHeadroomGate::over(Arc::new(
1358 ledger.with_fs_slack(fs_slack),
1359 )));
1360 match disk_gate.headroom() {
1361 Ok(room) => tracing::info!(
1362 root = %stores.root().display(),
1363 free = %gunnar_ssh::headroom::human(room.free),
1364 floor = %gunnar_ssh::headroom::human(room.floor),
1365 disk_floor = %config.disk_floor,
1366 has_room = room.has_room(),
1367 "disk headroom: pushes and Repos.Create are refused under the floor"
1368 ),
1369 Err(err) => tracing::error!(
1370 root = %stores.root().display(),
1371 %err,
1372 "disk headroom could not be measured; every push and Repos.Create will be \
1373 refused until it can be"
1374 ),
1375 }
1376 // The front doors' control: the gate CHAIN, disk first, then the plan.
1377 let front_control = || {
1378 front_control(
1379 &stores,
1380 config.anonymous.clone(),
1381 Arc::clone(&disk_gate),
1382 #[cfg(feature = "metered")]
1383 entitlement.as_ref(),
1384 )
1385 };
1386 #[cfg(feature = "http")]
1387 let http_lfs: Arc<dyn gunnar_ssh::seam::LfsTransport> = Arc::clone(&transport) as _;
1388 // **Repository settings over the forge's door** (owner, 2026-10-14):
1389 // `gunnar-repo …` on this listener, answered by the control plane's own
1390 // `Repos` handlers once the control plane exists (set further down);
1391 // until then — or on `--no-grpc` — the verb says it is not answered here.
1392 let repo_door_ctx: Arc<std::sync::OnceLock<gunnar_grpc::service::ServiceContext>> =
1393 Arc::new(std::sync::OnceLock::new());
1394 let ssh = GunnarSsh::new(ssh_config, front_control(), transport)
1395 // The boot window (`crate::readiness`): while the start-up disk
1396 // test is saturating /data, a git transfer is refused with
1397 // `starting` and EX_TEMPFAIL rather than served an open that
1398 // fails and reads as a missing repository.
1399 .with_readiness(crate::readiness::flag())
1400 // The admission gate is the ONE door all three git fronts ask, so
1401 // it is where a refusal is charged to the tenant that met it. See
1402 // `GunnarSsh::with_flow`.
1403 .with_flow(Arc::clone(stores.flow()))
1404 .with_metrics(Arc::clone(&metrics));
1405 // **The sealed box's tunnel-address door** (T17): `gunnar-tunnel-address`
1406 // over this listener, server-wide Admin, answered by the SAME function
1407 // `Bootstrap.AssignTunnelAddress` and `Admin.SetTunnelAddress` call. The
1408 // listener already held the caller to server Admin, so `may_write`.
1409 #[cfg(feature = "wg-client")]
1410 let ssh = {
1411 let root = stores.root().to_path_buf();
1412 ssh.with_tunnel_address_door(Arc::new(move |address: &str| {
1413 crate::appliance::tunnel_address_answer(&root, address, true)
1414 .map(|answer| crate::appliance::render_tunnel_answer(&answer))
1415 .map_err(|err| format!("{err}"))
1416 }))
1417 };
1418 let ssh = {
1419 let cell = Arc::clone(&repo_door_ctx);
1420 ssh.with_repo_door(Arc::new(move |ask: gunnar_ssh::RepoAsk| {
1421 let cell = Arc::clone(&cell);
1422 Box::pin(async move {
1423 let Some(ctx) = cell.get() else {
1424 return Err(
1425 "this server answers repository settings only on its control plane"
1426 .to_owned(),
1427 );
1428 };
1429 let setting = match ask.action {
1430 gunnar_ssh::RepoAction::Show => gunnar_grpc::repo_door::RepoSetting::Show,
1431 gunnar_ssh::RepoAction::Visibility(public) => {
1432 gunnar_grpc::repo_door::RepoSetting::Visibility(public)
1433 }
1434 gunnar_ssh::RepoAction::DefaultBranch(b) => {
1435 gunnar_grpc::repo_door::RepoSetting::DefaultBranch(b)
1436 }
1437 gunnar_ssh::RepoAction::Create { sha256 } => {
1438 gunnar_grpc::repo_door::RepoSetting::Create(sha256)
1439 }
1440 };
1441 gunnar_grpc::repo_door::ask(
1442 ctx,
1443 &ask.principal,
1444 &ask.fingerprint,
1445 ask.repo.namespace(),
1446 ask.repo.name(),
1447 setting,
1448 )
1449 .await
1450 .map(|repo| gunnar_grpc::repo_door::render(&repo))
1451 .map_err(|status| status.message().to_owned())
1452 })
1453 }))
1454 };
1455 // **--wg-only** (owner 2026-10-04): a connection at the utility address
1456 // — where the control plane is bound — is refused before a byte; the
1457 // public's git clients and the tunnel are served as ever.
1458 let ssh = if config.wg_only && !config.no_grpc {
1459 let utility = resolved_grpc_bind.ip();
1460 tracing::warn!(%utility, "ssh: --wg-only — a connection at the utility address is refused; boxes speak over the tunnel");
1461 ssh.with_refused_local(Arc::new(move |local: std::net::IpAddr| local == utility))
1462 } else {
1463 ssh
1464 };
1465 let ssh = ssh
1466 .bind(russh_key)
1467 .await
1468 .with_context(|| format!("binding the SSH listener on {}", config.ssh_bind))?;
1469 // The anonymous setting is on the *bind* line rather than in a separate
1470 // one, so an operator reading a boot log sees what this listener will
1471 // serve strangers in the same place they see where it is listening.
1472 tracing::info!(
1473 addr = %ssh.local_addr(),
1474 anonymous_read = config.allow_anonymous,
1475 "gunnar ssh listener: git-upload-pack, git-receive-pack; publickey, plus \
1476 anonymous read of public repositories when enabled"
1477 );
1478
1479 // ── the smart-HTTP listener ────────────────────────────────────────
1480 //
1481 // The SAME `Dispatch` the SSH listener authorises with, the SAME
1482 // `WireTransport`, the SAME counters. That is the whole point of the
1483 // crate: a second transport, not a second server. If this ever needed
1484 // its own control plane or its own transport, HTTP and SSH would be
1485 // able to disagree about who may push.
1486 #[cfg(feature = "http")]
1487 let http = match config.http_bind {
1488 None => None,
1489 Some(bind) => {
1490 // The TLS identity, resolved BEFORE binding: a `--http-tls-*`
1491 // that cannot produce one must be "the server refused to
1492 // start", never a listener that quietly came up cleartext —
1493 // that silent fallback would be the whole defect the flag
1494 // exists to prevent.
1495 let tls = match &config.http_tls {
1496 None => None,
1497 Some(HttpTls::Identity(identity)) => Some(identity.clone()),
1498 Some(HttpTls::SelfMinted) => {
1499 let pki = stores.pki().context(
1500 "--http-tls-self needs the secret store; no PKI is available",
1501 )?;
1502 let issuer = pki
1503 .issuer()
1504 .context("reading the issuer for --http-tls-self")?
1505 .context(
1506 "--http-tls-self needs an issuer and none exists; the first-start \
1507 mint above must have failed — its warning says why",
1508 )?;
1509 // The names a client may dial this listener by. rcgen
1510 // types each SAN as IP or DNS on its own. Loopback is
1511 // always present so `git clone https://127.0.0.1/…`
1512 // works out of the box; a concrete bind address joins
1513 // it. An operator with real DNS names brings their own
1514 // certificate — that is what `--http-tls-cert` is for.
1515 let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
1516 let ip = bind.ip();
1517 if !ip.is_unspecified() && !ip.is_loopback() {
1518 hosts.push(ip.to_string());
1519 }
1520 let issued = pki
1521 .issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
1522 .context("minting the smart-HTTP server certificate")?;
1523 let key_pem = issued.key_pem.clone().context(
1524 "the issuer minted a server certificate but returned no key",
1525 )?;
1526 tracing::info!(
1527 fingerprint = %issued.fingerprint,
1528 sans = ?hosts,
1529 "smart-HTTP TLS: server certificate minted from this server's own \
1530 issuer; clients pin the issuer (`gunnar issuer show --pem`)"
1531 );
1532 Some(gunnar_http::TlsIdentity {
1533 // Leaf first, issuer after, so a pinning client can
1534 // build the chain from the handshake alone.
1535 cert_chain_pem: format!("{}{}", issued.cert_pem, issuer.cert_pem),
1536 key_pem,
1537 })
1538 }
1539 };
1540 let http_config = gunnar_http::HttpConfig {
1541 listen: bind,
1542 // One field, both listeners. See `ServeConfig::allow_anonymous`.
1543 allow_anonymous: config.allow_anonymous,
1544 cleartext: config.http_cleartext,
1545 tls,
1546 ..gunnar_http::HttpConfig::default()
1547 };
1548 let bound = gunnar_http::GunnarHttp::new(
1549 http_config,
1550 gunnar_ssh::server::Dispatch::new(front_control()),
1551 Arc::clone(&http_transport),
1552 )
1553 .with_metrics(Arc::clone(&metrics))
1554 // ── ONE admission gate for BOTH front doors ─────────────────
1555 //
1556 // The SSH listener's own gate, handed in rather than rebuilt,
1557 // so a git transfer over smart HTTP counts against the same
1558 // server-wide limit, the same per-principal share and the same
1559 // memory budget as one over SSH. Before this the HTTP front had
1560 // no gate at all (measured: 64 clients, 64 slots, nothing
1561 // refused); two independent gates would have been the next
1562 // worst thing, doubling the appliance's real ceiling in
1563 // silence. This is the same-writer fix the crate's own comment
1564 // above demands: "a second transport, not a second server."
1565 .with_admission(ssh.admission_gate())
1566 .with_lfs(Arc::clone(&http_lfs))
1567 // The registry rides the same listener as git, on the same
1568 // credentials: one port, one identity table, one place an
1569 // operator opens in a firewall.
1570 .with_registry(Arc::new(crate::registry::StoreRegistry::new(Arc::clone(
1571 &stores,
1572 ))))
1573 .bind()
1574 .await
1575 .with_context(|| format!("binding the HTTP listener on {bind}"))?;
1576 let addr = bound.local_addr();
1577 let tls_on = bound.tls();
1578 tracing::info!(
1579 %addr,
1580 anonymous_read = config.allow_anonymous,
1581 cleartext_credentials = ?config.http_cleartext,
1582 tls = tls_on,
1583 "gunnar smart-http listener: info/refs, git-upload-pack, git-receive-pack, \
1584 info/lfs/objects (batch, get, put, verify)"
1585 );
1586 // A32-14, stays a task: a listener; its handle is kept and it ends at the server's stop.
1587 Some((tokio::spawn(bound.serve()), addr, tls_on))
1588 }
1589 };
1590
1591 // ── the scrape endpoint ────────────────────────────────────────────
1592 //
1593 // The SAME `Arc<ServerMetrics>` the heartbeat reads and both transports
1594 // write. Two readings of one set of counters, never two sets: a scrape
1595 // that disagreed with the heartbeat would leave an operator with no way
1596 // to decide which one to believe during the one hour they need either.
1597 //
1598 // On the GIT runtime and not the control plane's: the whole point of
1599 // this endpoint is that it still answers when the control plane is what
1600 // is wedged, and putting it on the pool that starves would have made it
1601 // useless in exactly that case.
1602 let scrape = match config.metrics_bind {
1603 None => None,
1604 Some(bind) => {
1605 let bound = crate::scrape::bind(bind, Arc::clone(&metrics))
1606 .await
1607 .with_context(|| format!("binding the metrics listener on {bind}"))?;
1608 let addr = bound.local_addr();
1609 // A32-14, stays a task: a listener; its handle is kept and it ends at the server's stop.
1610 Some((tokio::spawn(bound.serve()), addr))
1611 }
1612 };
1613
1614 // ── the gRPC control plane, on a runtime of its own ────────────────
1615 // ── replication slot: built EMPTY, and built EARLY ─────────────────
1616 //
1617 // Before the control plane, because the control plane takes a clone of
1618 // it: `Mirror.SetReplication` has to work on a box that has never had a
1619 // target and holds no repositories yet, which is every box on its first
1620 // boot. And it captures THIS runtime — the git one — because a blocking
1621 // send-pack belongs here and not on the control plane an operator is
1622 // using while the box is in trouble.
1623 #[cfg(feature = "replication")]
1624 let replication = crate::mirror::ReplicationSlot::new(
1625 Arc::clone(&stores),
1626 Arc::clone(&bus),
1627 tokio::runtime::Handle::current(),
1628 );
1629
1630 // ── the outbound webhook slot: same shape, same reason ─────────────
1631 //
1632 // Built empty and early, for the same reason the replication slot is:
1633 // the control plane takes a clone of it so `Webhooks.AddWebhook` works
1634 // on a box that has never had an endpoint. Its workers do async POSTs,
1635 // not blocking send-packs, but sharing the git runtime keeps them off
1636 // the control plane an operator uses while the box is in trouble.
1637 #[cfg(feature = "webhooks")]
1638 let webhook = crate::webhook::WebhookSlot::new(
1639 Arc::clone(&stores),
1640 Arc::clone(&bus),
1641 tokio::runtime::Handle::current(),
1642 );
1643
1644 // The control plane's copy of the replication view, taken while the
1645 // plane is built and filled below once the replicator exists.
1646 let mut control_view: Option<crate::replication_view::ReplicationView> = None;
1647 #[cfg(feature = "replication")]
1648 let mut control_for_repair: Option<Arc<dyn gunnar_grpc::ControlPlane>> = None;
1649 // The SAME grow machine `Admin.GrowStart` drives: a pulled `grow_go_ahead` (G16-d).
1650 #[cfg(feature = "replication")]
1651 let mut grow_for_repair: Option<Arc<crate::grow::GrowMachine>> = None;
1652 let (grpc, grpc_shutdown, grpc_endpoint, control_runtime) = if config.no_grpc {
1653 tracing::warn!(
1654 "the gRPC control plane is disabled; repositories, keys and grants can only be \
1655 administered from this host's command line"
1656 );
1657 (None, None, None, ControlRuntime::default())
1658 } else {
1659 let workers = config
1660 .control_plane_workers
1661 .unwrap_or_else(control_plane_workers);
1662 // **The calls on one runtime, the connections on another**
1663 // (R5.1-G). The handler runtime is held by a guard from the moment
1664 // it exists, so the `?` below cannot drop a bare runtime here (see
1665 // `ControlRuntime`).
1666 let mut handler_runtime = ControlRuntime::new(
1667 tokio::runtime::Builder::new_multi_thread()
1668 .worker_threads(workers)
1669 .thread_name("gunnar-control-call")
1670 .enable_all()
1671 .build()
1672 .context("building the control plane's handler runtime")?,
1673 );
1674 let handlers = handler_runtime
1675 .handle()
1676 .context("the control plane's handler runtime was gone as soon as it was built")?;
1677 let control_runtime = tokio::runtime::Builder::new_multi_thread()
1678 .worker_threads(CONTROL_CONNECTION_WORKERS)
1679 .thread_name("gunnar-control")
1680 .enable_all()
1681 .build()
1682 .context("building the control plane's runtime")?;
1683
1684 let grpc_bind = resolved_grpc_bind;
1685 let grpc_bind_device = config.grpc_bind_device.clone();
1686 let web_relying_parties = config
1687 .web_relying_parties
1688 .iter()
1689 .map(|party| {
1690 party
1691 .parse::<gunnar_grpc::webauthn::RelyingParty>()
1692 .map_err(|why| anyhow::anyhow!("--web-relying-party: {why}"))
1693 })
1694 .collect::<Result<Vec<_>>>()?;
1695 let ssh_hint = ssh.local_addr().to_string();
1696 // **ONE twin heartbeat registry, shared by the gRPC plane that
1697 // RECORDS a twin's poll and the drain that READS what it applied.**
1698 //
1699 // It used to be created inside `ServiceContext` and owned only
1700 // there, so the grow world could not see it — and the drain proved
1701 // itself instead by pushing to the replication roster, which for a
1702 // twin can never succeed (a twin serves no git). Built here so both
1703 // halves hold the same `Arc`; two registries would be two answers
1704 // to "what has the twin applied", and the one in front of the
1705 // unmount would be the empty one.
1706 let heartbeats = Arc::new(gunnar_grpc::heartbeat::Registry::default());
1707 // ── the runtime data-set growth (`crate::grow`) ────────────────
1708 //
1709 // Over the ONE admission gate the fronts share (the drain pauses
1710 // it) and these stores (the identity handle, the close/reopen).
1711 // With `grow-runtime` the world is the box; without it the machine
1712 // still answers `GrowStatus` honestly and refuses `GrowStart` by
1713 // name.
1714 #[cfg(feature = "grow-runtime")]
1715 let grow_world: Arc<dyn crate::grow::World> = Arc::new(
1716 crate::grow::world::BoxWorld::new(ssh.admission_gate(), Arc::clone(&stores))
1717 // So the drain can prove the standbys are caught up once
1718 // the gate is shut and nothing is in flight — the only
1719 // moment at which "everything is drained to the twin" is
1720 // a true sentence. Without this the world falls back to
1721 // `NoStandby` and a growth would proceed having proved
1722 // nothing about a twin that exists.
1723 .with_replication(Arc::clone(&replication))
1724 // The twin's OWN reported position — the only thing on
1725 // this box that says whether it holds the set.
1726 .with_heartbeats(Arc::clone(&heartbeats)),
1727 );
1728 #[cfg(not(feature = "grow-runtime"))]
1729 let grow_world: Arc<dyn crate::grow::World> = Arc::new(crate::grow::world::NoEngine);
1730 // **The go-ahead policy is the metered half's, and only the metered
1731 // half's.** AUDITED 2026-09-18: this line read
1732 // `GrowMachine::new(grow_world)` on every build, which installs
1733 // `OperatorPolicy` — and that policy refuses every signer but
1734 // `operator` by name. monetize signs its go-ahead `signer:
1735 // "monetize"`, so its orchestrator reached `phase::STARTING`,
1736 // called `Admin.GrowStart` and was refused AFTER it had reserved
1737 // the spend: front-driven growth was dead at step 3 and had never
1738 // worked. `MonetizePolicy` verifies that signature under the key
1739 // the box already holds from `--monetize-pubkey` (the same one
1740 // `EntitlementSlot::open` above was given), and it takes an
1741 // `operator` go-ahead exactly as before. `cfg`, not a runtime
1742 // branch: a community binary has no monetize verifier in it at all.
1743 #[cfg(feature = "metered")]
1744 let grow = match &config.metered {
1745 Some(metered) => {
1746 // Keyless when the key is fed and has not arrived: it is then
1747 // applied here in place too (`feed::on_monetize_key`).
1748 let policy = Arc::new(match &metered.monetize_pubkey {
1749 Some(key) => crate::grow::MonetizePolicy::new(key)
1750 .map_err(|e| anyhow::anyhow!("the growth go-ahead policy: {e}"))?,
1751 None => crate::grow::MonetizePolicy::awaiting(),
1752 });
1753 let taker = Arc::clone(&policy);
1754 crate::feed::on_monetize_key(
1755 "growth go-ahead policy",
1756 Box::new(move |k| taker.set_key(k)),
1757 );
1758 crate::grow::GrowMachine::with_policy(grow_world, policy)
1759 }
1760 // A metered binary started without `--monetize-pubkey` serves
1761 // Free (the warning three hundred lines up); it has no key to
1762 // verify a `monetize` go-ahead with either, so it keeps the
1763 // operator door and refuses monetize by name rather than
1764 // pretending to a check it cannot make.
1765 None => crate::grow::GrowMachine::new(grow_world),
1766 };
1767 #[cfg(not(feature = "metered"))]
1768 let grow = crate::grow::GrowMachine::new(grow_world);
1769 #[cfg(feature = "replication")]
1770 {
1771 grow_for_repair = Some(Arc::clone(&grow));
1772 }
1773 let mut plane = CoreControlPlane::new(Arc::clone(&stores))
1774 // `Bootstrap.Restore`'s keys from the medium; the fed ones are read per ask.
1775 .with_restore_keys(restore_keys.clone())
1776 // `Repos.Create` asks the SAME gate the front doors ask (one implementation):
1777 // a disk that refuses a push refuses a new repository with the
1778 // same line.
1779 .with_headroom(Arc::clone(&disk_gate))
1780 .with_grow(grow)
1781 // The in-flight count `Mirror.FlushReplication` reports (T11).
1782 .with_admission(ssh.admission_gate())
1783 // The SAME bus the SSH front door publishes a push onto, so a
1784 // `MirrorSynced` and a `RefUpdated` reach one `Events.Watch`
1785 // stream. The plane publishes it from the mirror queue's worker
1786 // when a sync completes — `Mirror.Sync` itself only enqueues,
1787 // and an event announcing a fetch that has not happened would
1788 // be worse than no event.
1789 .with_bus(Arc::clone(&bus))
1790 // `Bootstrap.Status` names the role before the join (a twin has no banner).
1791 .with_twin_role(config.twin_role)
1792 // **`serving_git` is a reading of THIS socket.** Both addresses,
1793 // so the status door can say what was asked for and what was
1794 // taken: on a twin those differ, because the branch above binds
1795 // loopback:0 whatever `--ssh-bind` named, and a box that reports
1796 // `git serving` while its :2222 sends an RST is a health line
1797 // that cannot fail.
1798 .with_git_listener(crate::readiness::GitListener {
1799 asked: config.ssh_bind,
1800 bound: ssh.local_addr(),
1801 });
1802 // The plane publishes into the SAME view the band reads, so
1803 // `Admin.Stats` and the console can never disagree about whether a
1804 // standby is taking bytes (one implementation).
1805 // The WRITE side. `replication_view` just below is the READ side
1806 // one slot, two halves, and neither is the other's twin —
1807 // a `SetReplication` changes what is running, a `Stats` reads what
1808 // it did.
1809 #[cfg(feature = "replication")]
1810 {
1811 plane = plane.with_replication(Arc::clone(&replication));
1812 }
1813 #[cfg(feature = "webhooks")]
1814 {
1815 plane = plane.with_webhooks(Arc::clone(&webhook));
1816 }
1817 #[cfg(feature = "metered")]
1818 if let Some(slot) = &entitlement {
1819 plane = plane.with_entitlement(Arc::clone(slot));
1820 }
1821 let plane_view = plane.replication_view();
1822 if let Some(bytes) = config.max_object_bytes {
1823 plane = plane.with_max_object_bytes(bytes);
1824 }
1825 // **An import the last process did not live to finish is settled and
1826 // resumed** (import-big, 2026-10-26), on the control runtime the
1827 // mirror queue's lane lives on. A twin originates no sync.
1828 if !config.twin_role {
1829 let _on = control_runtime.enter();
1830 let resumed = plane.resume_interrupted_mirrors();
1831 if resumed > 0 {
1832 tracing::info!(target: "gunnar::git", resumed, "mirror syncs interrupted by the last stop were resumed");
1833 }
1834 }
1835 // The trust root, resolved BEFORE the listener is spawned, for the
1836 // reason `--http-tls-*` is resolved before its bind: a control
1837 // plane that was asked for client certificates and could not mint
1838 // its own identity must be a startup error naming the cause, never
1839 // a listener that came up accepting anyone.
1840 let grpc_trust = match config.grpc_tls {
1841 None => None,
1842 Some(GrpcTls::SelfMinted) => {
1843 let pki = stores
1844 .pki()
1845 .context("--grpc-tls-self needs the secret store; no PKI is available")?;
1846 let issuer = pki
1847 .issuer()
1848 .context("reading the issuer for --grpc-tls-self")?
1849 .context(
1850 "--grpc-tls-self needs an issuer and none exists; the first-start \
1851 mint above must have failed — its warning says why",
1852 )?;
1853 // The same SAN reasoning as the HTTP listener's: loopback
1854 // always, plus a concrete bind address when there is one.
1855 let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
1856 let ip = grpc_bind.ip();
1857 if !ip.is_unspecified() && !ip.is_loopback() {
1858 hosts.push(ip.to_string());
1859 }
1860 let issued = pki
1861 .issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
1862 .context("minting the gRPC server certificate")?;
1863 let key_pem = issued
1864 .key_pem
1865 .clone()
1866 .context("the issuer minted a server certificate but returned no key")?;
1867 tracing::info!(
1868 fingerprint = %issued.fingerprint,
1869 sans = ?hosts,
1870 "gRPC control plane: mutual TLS on. Clients must present a certificate \
1871 this server's issuer signed AND registered (`gunnar control bootstrap`, \
1872 `gunnar key add`)"
1873 );
1874 Some(gunnar_grpc::Trust {
1875 // The issuer is both halves here: the CA every client
1876 // certificate must chain to, and the second link of
1877 // the chain this listener presents. Beside it, the
1878 // CENTRAL monetize's attach CA when this estate
1879 // attached from another cloud (f-grpc-tls): the
1880 // central dials in with a client identity under it.
1881 #[cfg(feature = "wg-client")]
1882 client_ca: crate::fed_principals::grpc_client_cas(
1883 &issuer.cert_pem,
1884 stores.root(),
1885 )
1886 .into_bytes(),
1887 // No fed attach CA without the installer's feed (wg-client).
1888 #[cfg(not(feature = "wg-client"))]
1889 client_ca: issuer.cert_pem.clone().into_bytes(),
1890 cert: format!("{}{}", issued.cert_pem, issuer.cert_pem).into_bytes(),
1891 key: key_pem.into_bytes(),
1892 })
1893 }
1894 };
1895 // Taken from the trust root itself rather than from
1896 // `config.grpc_tls`, so the field `Admin.Health` reports is the
1897 // one the listener is actually built with.
1898 let grpc_mutual_tls = grpc_trust.is_some();
1899
1900 // ── the MUTUAL REPLICATION LISTENER (lane T9) ──────────────────────
1901 //
1902 // The door a twin follows this box through. Resolved HERE, before
1903 // anything is spawned, for the reason the main trust root is: a box
1904 // that should serve its twin and cannot mint the identity to do it
1905 // must fail to start NAMING the cause, not come up with a twin that
1906 // retries into nothing for ever. `grpc_replication_bind` unset derives
1907 // it from the main bind (off loopback ⇒ port + 1), which is what lets
1908 // the one medium carry no flag.
1909 let replication_root = stores.root().to_path_buf();
1910 let replication = {
1911 // The ONE derivation (`client_signer::replication_address`), so the
1912 // operator side that dials it cannot drift from what is bound here.
1913 let derived = (!grpc_bind.ip().is_loopback() && grpc_bind.port() != 0)
1914 .then(|| {
1915 gunnar_grpc::client_signer::replication_address(&grpc_bind.to_string()).ok()
1916 })
1917 .flatten()
1918 .and_then(|at| at.parse::<SocketAddr>().ok());
1919 // **Where it binds** (owner 2026-09-22, amended 2026-09-24): on the
1920 // derived (eth1) address for the process lifetime, and ALSO on
1921 // `wg0`'s address while the tunnel is up (`replication_door`); an
1922 // explicit `--grpc-replication-bind` is never widened.
1923 let explicit = config.grpc_replication_bind.is_some();
1924 match config.grpc_replication_bind.or(derived) {
1925 None => None,
1926 Some(base) => {
1927 let bind = base;
1928 let pki = stores.pki().with_context(|| {
1929 format!("the mutual replication listener on {bind} needs the secret store; no PKI is available")
1930 })?;
1931 let issuer = pki.issuer().context("reading the issuer for the replication listener")?.with_context(|| {
1932 format!(
1933 "the mutual replication listener on {bind} needs this box's issuer and there is \
1934 none — the first-start mint above must have failed, and its warning says why. \
1935 Without it no twin of this box can ever follow it."
1936 )
1937 })?;
1938 let mut hosts = vec![
1939 gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
1940 "localhost".to_owned(),
1941 "127.0.0.1".to_owned(),
1942 ];
1943 for ip in [bind.ip(), base.ip()] {
1944 if !ip.is_unspecified()
1945 && !ip.is_loopback()
1946 && !hosts.contains(&ip.to_string())
1947 {
1948 hosts.push(ip.to_string());
1949 }
1950 }
1951 // Minted at EVERY start, so a restart is a renewal and a
1952 // lapse needs a year of uninterrupted uptime.
1953 let issued = pki
1954 .issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
1955 .context("minting the replication listener's certificate")?;
1956 let key_pem = issued.key_pem.clone().context(
1957 "the issuer minted the replication certificate but returned no key",
1958 )?;
1959 Some((
1960 base,
1961 explicit,
1962 bind,
1963 gunnar_grpc::Trust {
1964 client_ca: issuer.cert_pem.clone().into_bytes(),
1965 cert: format!("{}{}", issued.cert_pem, issuer.cert_pem)
1966 .into_bytes(),
1967 key: key_pem.into_bytes(),
1968 },
1969 ))
1970 }
1971 }
1972 };
1973
1974 let control = Arc::new(plane);
1975 control_view = Some(plane_view);
1976 #[cfg(feature = "replication")]
1977 {
1978 control_for_repair = Some(control.clone());
1979 }
1980 let bus_for_grpc = Arc::clone(&bus);
1981 // ── ONE admission gate on ALL THREE front doors ────────────────
1982 //
1983 // `Repos.ArchiveOut` streams a whole repository's stored bytes down
1984 // the control-plane channel — a transfer in everything but name — so
1985 // it takes a permit from the SAME `Arc<Admission>` the SSH and HTTP
1986 // git fronts hold (`ssh.admission_gate()`), not a second gate (one
1987 // implementation). The *cheap* control RPCs get no gate on purpose: the control
1988 // plane runs on its own runtime, so their walks starve
1989 // nothing but themselves. Captured here and moved into the control
1990 // runtime's spawn below.
1991 let grpc_transfer_gate = crate::grpc_gate::control_transfer_gate(ssh.admission_gate());
1992 let browse_flag = stores.browse();
1993 // ── the durable audit trail (GUN-044) ──────────────────────────
1994 //
1995 // Opened HERE, on the startup path, because `AuditChain::open`
1996 // verifies the existing chain and an operator has to be told about
1997 // a broken one at the moment the server starts — not the first time
1998 // somebody runs a verification months later.
1999 //
2000 // A chain that will not open does NOT stop the server. The control
2001 // plane still authenticates, still authorises and still writes the
2002 // `tracing` record it always did; what it loses is the durable
2003 // copy, and it says so at `error!` with the reason. Refusing to
2004 // serve would turn a full disk — or a chain somebody has already
2005 // tampered with — into an outage, which hands whoever tampered
2006 // with it exactly what they wanted.
2007 let audit_ledger = {
2008 // **Off the stripe.** The trail records who changed what on
2009 // this box, which is exactly the question asked the morning
2010 // after the four striped data disks have gone — so a copy that
2011 // dies with them answers nobody. `ServerStores::audit_path`
2012 // adopts an existing chain from the control directory once and
2013 // falls back to it unchanged if the slow disk will not take it.
2014 let path = stores.audit_path();
2015 // ── and it must SURVIVE A RESTORE (lane T11, 2026-09-21) ───
2016 //
2017 // D6: `log/` was a per-box path, so the twin never held the
2018 // trail and `twin replay` brought a box back with no audit
2019 // history at all — silently, reporting success. A system whose
2020 // audit trail is destroyed by its own recovery procedure
2021 // cannot answer the only question that will be asked.
2022 //
2023 // So the primary's chain is opened FRAMED: every entry is an
2024 // `Append` transaction, the twin holds it, and a restored box
2025 // continues the dead primary's chain from its own tip.
2026 //
2027 // A twin — and a box mid-restore — originates nothing, so it
2028 // gets an ordinary chain at its OWN per-box path
2029 // (`audit_path` decides the name), and this branch does not
2030 // even try to frame it.
2031 let originates = !stores.is_twin() && !crate::twin::is_restoring(stores.root());
2032 let opened = if originates {
2033 match gunnar_core::AuditChain::open_framed(&path) {
2034 Ok(chain) => Ok(chain),
2035 // **Never silent.** An unframable chain is the D6 bug
2036 // coming back, so it is said in the words an operator
2037 // needs, and the server still writes a local trail.
2038 Err(err @ gunnar_core::AuditError::Unframed { .. }) => {
2039 tracing::error!(
2040 target: gunnar_grpc::AUDIT_TARGET,
2041 path = %path.display(),
2042 error = %err,
2043 "this box's audit trail is NOT replicated: it will be written \
2044 here and it will NOT survive a restore from the twin. Every \
2045 other class of state comes back; this one will not."
2046 );
2047 gunnar_core::AuditChain::open(&path)
2048 }
2049 Err(err) => Err(err),
2050 }
2051 } else {
2052 gunnar_core::AuditChain::open(&path)
2053 };
2054 match opened {
2055 Ok(chain) => Some(crate::grpc_gate::control_audit_ledger(Arc::new(chain))),
2056 Err(err) => {
2057 tracing::error!(
2058 target: gunnar_grpc::AUDIT_TARGET,
2059 path = %path.display(),
2060 error = %err,
2061 "the control-plane audit chain could not be opened; this server \
2062 will log its decisions and will NOT write them to a trail. If the \
2063 chain is reported BROKEN, the file has been edited or truncated \
2064 since gunnar wrote it."
2065 );
2066 None
2067 }
2068 }
2069 };
2070 let (tx, rx) = oneshot::channel();
2071 // The listener is bound INSIDE the control runtime. A tokio socket
2072 // registers with the I/O driver of the runtime that created it, so
2073 // one bound out here would keep the git runtime's driver in the
2074 // control plane's critical path, which is the exact coupling this
2075 // change exists to remove.
2076 let (bound_tx, bound_rx) = oneshot::channel::<Result<(SocketAddr, bool)>>();
2077 // The box's jera ledger (RA3-5): the job queue's jobs and the long duties are rows in
2078 // `<data>/jobs`, read by `nornir jobs`; one that cannot open is said, and they run unrecorded.
2079 if let Err(e) = gunnar_common::jobs::set_ledger(&stores.root().join("jobs")) {
2080 tracing::error!("ALERT jobs.ledger {e}: the jobs run, unrecorded");
2081 }
2082 let sweep_stores = Arc::clone(&stores);
2083 let activity_stores = Arc::clone(&stores);
2084 let wg_only = config.wg_only;
2085 // A32-14, stays a task: the control plane on its own runtime; its handle is kept and joined at the stop.
2086 let join = control_runtime.spawn(async move {
2087 // Bound WITH PATIENCE: a control plane on a tailnet or VPN
2088 // address is asked for before that interface exists on a cold
2089 // boot (builder, every boot until 2026-09-10: six `EADDRNOTAVAIL`
2090 // exits and six blind systemd restarts before the seventh try
2091 // found the address). The wait is bounded and only for that
2092 // one errno; everything else still fails at once.
2093 let listener = match bind_when_the_address_arrives(
2094 || bind_grpc_listener(grpc_bind, grpc_bind_device.as_deref()),
2095 GRPC_BIND_PATIENCE,
2096 GRPC_BIND_RETRY_EVERY,
2097 )
2098 .await
2099 .0
2100 .with_context(|| match &grpc_bind_device {
2101 Some(dev) => format!("binding the gRPC listener on {grpc_bind} (device {dev})"),
2102 None => format!("binding the gRPC listener on {grpc_bind}"),
2103 }) {
2104 Ok(listener) => listener,
2105 Err(err) => {
2106 // Report the bind failure to `start`, which is the only
2107 // place that can turn it into a startup error. Swallowing
2108 // it here would leave `gunnar serve` announcing a control
2109 // plane that never bound.
2110 let _ = bound_tx.send(Err(err));
2111 return Ok(());
2112 }
2113 };
2114 let addr = match listener
2115 .local_addr()
2116 .context("reading the gRPC local address")
2117 {
2118 Ok(addr) => addr,
2119 Err(err) => {
2120 let _ = bound_tx.send(Err(err));
2121 return Ok(());
2122 }
2123 };
2124 let server_config = ServerConfig::default()
2125 .with_bind(addr)
2126 .with_ssh_bind_hint(ssh_hint)
2127 .with_web_relying_parties(web_relying_parties)
2128 .with_web_user_verification(if config.web_require_user_verification {
2129 gunnar_grpc::config::UserVerification::Required
2130 } else {
2131 gunnar_grpc::config::UserVerification::Preferred
2132 })
2133 // Reported from the SAME value that decides whether the
2134 // listener is built with a trust root, three lines below
2135 // (`router_with_trust(…, grpc_trust.as_ref())`). Derived
2136 // from anything else it would be a second opinion about
2137 // whether the door is locked, and `Admin.Health`'s job
2138 // here is to be the answer rather than to remember one.
2139 .with_mutual_tls(grpc_mutual_tls);
2140 // Every start, on purpose: a hole nobody is reminded of is a hole
2141 // nobody fixes.
2142 gunnar_grpc::warn_authentication_disabled(&server_config);
2143 // Reported out rather than re-derived by the caller. `serve`
2144 // prints this on stdout, and a second reader of "is auth on"
2145 // is a second thing that can be wrong: the line it replaced
2146 // said `UNAUTHENTICATED in v1` unconditionally, and had
2147 // gone stale against a control plane that authenticates.
2148 let authenticated = server_config.authentication_enabled();
2149 let ctx = ServiceContext::new(control, bus_for_grpc, server_config)
2150 // The one "this estate sells room" flag the push gate reads too.
2151 .with_sells_room(Arc::clone(&sells_room))
2152 // Every call on the handler runtime, the connections on
2153 // this one (R5.1-G): no slow call can hold the worker that
2154 // answers a client's keepalive PING.
2155 .with_handler_runtime(handlers)
2156 // The transfer-shaped RPC's permit, from the git
2157 // fronts' own gate.
2158 .with_transfer_gate(grpc_transfer_gate)
2159 // The browse replica: the browse flag the mutator refusal reads —
2160 // the same `Arc` the replication slot flips.
2161 .with_browse_flag(browse_flag)
2162 // The SAME registry the drain reads (see where it is built).
2163 .with_heartbeats(Arc::clone(&heartbeats));
2164 // The hash-chained trail, when there is one; `tracing` alone
2165 // when there is not. Either way the gate's decisions also
2166 // stamp who last changed something (2026-10-21, the admin's
2167 // user list): `grpc_gate::activity_ledger`.
2168 let ctx = ctx.with_audit_ledger(crate::grpc_gate::activity_ledger(
2169 audit_ledger,
2170 Arc::clone(activity_stores.activity()),
2171 ));
2172 // **The identity core** (IDENTITY-HOST, 2026-10-27): edda's one
2173 // copy of sign-up, passkey login, invites, sessions and step-up,
2174 // over Gunnar's host; its records persist beside the control state.
2175 // Its records fail CLOSED (RA5-9): a document that does not read stops the
2176 // start by name; empty tables over it would revive the deactivated.
2177 let identity = match crate::identity::try_build(
2178 activity_stores.root(),
2179 Arc::clone(&ctx.control),
2180 &ctx.config,
2181 config.identity_root_site,
2182 ) {
2183 Ok(identity) => identity,
2184 Err(why) => {
2185 let _ = bound_tx.send(Err(anyhow::anyhow!("{why}")));
2186 return Ok(());
2187 }
2188 };
2189 activity_stores.set_identity(&identity);
2190 let ctx = ctx.with_identity(identity);
2191 // The forge door's `gunnar-repo` verb answers from here on.
2192 let _ = repo_door_ctx.set(ctx.clone());
2193 // **Company SSO's network**: the one way gunnar reaches an
2194 // account's identity provider. `GUNNAR_SSO_ALLOW_LOOPBACK_HTTP=1`
2195 // is the operator's test door (an IdP on this box's loopback,
2196 // over http) and nothing else.
2197 let sso_loopback = std::env::var("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP").is_ok_and(|v| v.trim() == "1");
2198 if sso_loopback {
2199 tracing::warn!("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP=1: an identity provider on this box's loopback is reachable over http (a test door)");
2200 }
2201 ctx.sso.set_allow_loopback_http(sso_loopback);
2202 // A32-18: the egress is `sso-egress`'s (every appliance row names it); a build
2203 // without it sets none, so an identity provider's call is refused by name.
2204 #[cfg(feature = "sso-egress")]
2205 ctx.sso.set_egress(std::sync::Arc::new(crate::sso_egress::UreqEgress::with_nets(
2206 sso_loopback,
2207 crate::sso_egress::system_lookup(),
2208 ctx.sso.private_nets_handle(),
2209 )));
2210 #[cfg(not(feature = "sso-egress"))]
2211 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");
2212 // **The operator's switch** (A34-16, default = AIRGAP): the wired egress is
2213 // handed out only while the feed says `sso_egress=on` (or, on a box fed
2214 // nothing, `GUNNAR_SSO_EGRESS=on`); applied now and on every feed after.
2215 {
2216 let env = std::env::var("GUNNAR_SSO_EGRESS").ok();
2217 let on = crate::feed::sso_egress_on(crate::feed::Fed::current().as_ref(), env.as_deref());
2218 ctx.sso.set_egress_on(on);
2219 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" });
2220 let sso = std::sync::Arc::clone(&ctx.sso);
2221 // The resolver a domain's DNS proof is read from (A34-5): the feed's, or
2222 // `GUNNAR_SSO_DNS_RESOLVER` where the feed names none.
2223 let resolver_env = std::env::var("GUNNAR_SSO_DNS_RESOLVER").ok();
2224 if let Some(resolver) = crate::feed::sso_dns_resolver(crate::feed::Fed::current().as_ref(), resolver_env.as_deref()) {
2225 ctx.sso.set_dns_resolver(&resolver);
2226 }
2227 nordisk_pairing::feedbus::on_published("sso: the operator's egress switch", move |feed| {
2228 let fed = crate::feed::Fed::from_feed(feed);
2229 // The resolver, live.
2230 sso.set_dns_resolver(crate::feed::sso_dns_resolver(Some(&fed), resolver_env.as_deref()).as_deref().unwrap_or(""));
2231 let on = crate::feed::sso_egress_on(Some(&fed), env.as_deref());
2232 if on != sso.egress_on() {
2233 tracing::info!(target: "gunnar::audit", event = "sso.egress", on, "company SSO's network turned {}", if on { "ON" } else { "off" });
2234 sso.set_egress_on(on);
2235 }
2236 });
2237 }
2238 // **The operator's internal networks a company IdP may live
2239 // on** (owner 2026-10-06): the installer's feed, applied now and
2240 // on every feed after (LIVE); a list gunnar refuses is said and
2241 // applied as none — today's guard.
2242 {
2243 let sso = std::sync::Arc::clone(&ctx.sso);
2244 nordisk_pairing::feedbus::on_published(
2245 "sso: the operator's identity-provider networks",
2246 move |feed| {
2247 let nets = match crate::feed::sso_private_nets(Some(&crate::feed::Fed::from_feed(feed))) {
2248 Ok(nets) => nets,
2249 Err(why) => {
2250 tracing::warn!(target: "gunnar::audit", event = "sso.private_nets_refused", reason = %why, "the fed identity-provider networks were refused; none are admitted");
2251 gunnar_grpc::sso::PrivateNets::default()
2252 }
2253 };
2254 if nets != sso.private_nets() {
2255 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() });
2256 sso.set_private_nets(nets);
2257 }
2258 },
2259 );
2260 }
2261 // **The transaction log's default retention** (h-txlog-2x): a PRIMARY
2262 // retains its log behind its slowest REPORTING twin, minus a window, every
2263 // ten minutes — never with no twin, never past a silent or unheard one, and
2264 // never on a twin (its log is its only copy; the pass says so and does
2265 // nothing). `GUNNAR_TXLOG_RETAIN=off` turns it off.
2266 // A32-14, stays a task: a supervised async loop (a panicked pass is said) on the server's runtime; it ends with it.
2267 tokio::spawn(gunnar_grpc::retention::run(ctx.clone()));
2268 // **The account audit logs' retention** (2026-10-05): a year by
2269 // default, swept every six hours on a primary.
2270 // A jera job (A32-14) that ends with the stores; one that cannot start is said.
2271 if let Err(e) = crate::account_audit_sweep::start(&sweep_stores, crate::account_audit_sweep::EVERY) {
2272 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);
2273 }
2274 // The replication listener: bound BEFORE the address is reported,
2275 // so a port that will not bind is a startup error and not a twin
2276 // that retries into a closed door. The same context — the same
2277 // gate, grants and audit trail — on a door that demands a
2278 // certificate at the handshake.
2279 // One stop for the replication door, whichever address it is on
2280 // (it may rebind when the tunnel comes up at runtime).
2281 let (replication_stop_tx, replication_stop_rx) = tokio::sync::watch::channel(false);
2282 match replication {
2283 Some((base, explicit, bind, trust)) => {
2284 let listener = match tokio::net::TcpListener::bind(bind).await {
2285 Ok(listener) => listener,
2286 Err(err) => {
2287 let _ = bound_tx.send(Err(anyhow::Error::new(err).context(format!(
2288 "binding the mutual replication listener on {bind} (the door this box's twin follows it through)"
2289 ))));
2290 return Ok(());
2291 }
2292 };
2293 let bound = listener.local_addr().unwrap_or(base);
2294 let recorded = if wg_only { record_replication_door_wg_only(&replication_root, bound.port(), None) } else { record_replication_door(&replication_root, bound, None) };
2295 if let Err(err) = recorded {
2296 let _ = bound_tx.send(Err(err));
2297 return Ok(());
2298 }
2299 // Built once up front so a broken trust is a STARTUP error;
2300 // every rebind builds its own from the same trust.
2301 if let Err(err) = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, Some(&trust)) {
2302 let _ = bound_tx.send(Err(err.context("configuring the mutual replication listener")));
2303 return Ok(());
2304 }
2305 tracing::info!(
2306 %bind, suite = gunnar_grpc::preferred_bulk_suite(),
2307 "mutual replication listener up: TLS 1.3, a client certificate from this box's issuer required"
2308 );
2309 // A32-14, stays a task: the replication listener's accept loop, ended by the server's stop.
2310 tokio::spawn(replication_door(
2311 listener,
2312 base,
2313 explicit,
2314 wg_only,
2315 ctx.clone(),
2316 grpc_request_id,
2317 trust,
2318 replication_root.clone(),
2319 replication_stop_rx,
2320 ));
2321 }
2322 None => {
2323 if let Err(err) = crate::twin_tls::record_replication_port(&replication_root, None) {
2324 let _ = bound_tx.send(Err(err));
2325 return Ok(());
2326 }
2327 }
2328 }
2329 if bound_tx.send(Ok((addr, authenticated))).is_err() {
2330 // `start` gave up on us; do not leave a listener running.
2331 return Ok(());
2332 }
2333 if wg_only {
2334 // **--wg-only** (owner 2026-10-04): the bound — utility —
2335 // listener serves nothing; the control plane answers on the
2336 // tunnel's address only, following it (`control_doors`).
2337 gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref())
2338 .context("configuring mutual TLS on the gRPC control plane")?;
2339 let (control_stop_tx, control_stop_rx) = tokio::sync::watch::channel(false);
2340 let serve_control = move |l: tokio::net::TcpListener, mut stop: tokio::sync::watch::Receiver<bool>| {
2341 let router = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref());
2342 async move {
2343 let router = match router {
2344 Ok(r) => r,
2345 Err(err) => {
2346 tracing::error!(%err, "the control plane's tunnel door could not be configured");
2347 return;
2348 }
2349 };
2350 let at = l.local_addr().map(|a| a.to_string()).unwrap_or_default();
2351 if let Err(err) = router
2352 .serve_with_incoming_shutdown(tokio_stream::wrappers::TcpListenerStream::new(l), async move {
2353 let _ = stop.wait_for(|s| *s).await;
2354 })
2355 .await
2356 {
2357 tracing::error!(%err, %at, "a control plane door stopped");
2358 }
2359 }
2360 };
2361 // A32-14, stays a task: the control doors' accept loop; its handle is kept and it stops at the stop signal.
2362 let doors = tokio::spawn(control_doors(listener, addr, true, crate::tunnel_addr::subscribe(), control_stop_rx, serve_control));
2363 let _ = rx.await;
2364 // One stop for every door.
2365 let _ = replication_stop_tx.send(true);
2366 let _ = control_stop_tx.send(true);
2367 let _ = doors.await;
2368 Ok(())
2369 } else {
2370 gunnar_grpc::router_with_trust(ctx, grpc_request_id, grpc_trust.as_ref())
2371 .context("configuring mutual TLS on the gRPC control plane")?
2372 .serve_with_incoming_shutdown(
2373 tokio_stream::wrappers::TcpListenerStream::new(listener),
2374 async {
2375 let _ = rx.await;
2376 // One stop for both doors.
2377 let _ = replication_stop_tx.send(true);
2378 },
2379 )
2380 .await
2381 .context("the gRPC control plane stopped")
2382 }
2383 });
2384
2385 let (addr, grpc_authenticated) = match bound_rx.await {
2386 Ok(Ok(pair)) => pair,
2387 Ok(Err(err)) => {
2388 control_runtime.shutdown_background();
2389 return Err(err);
2390 }
2391 Err(_) => {
2392 control_runtime.shutdown_background();
2393 bail!(
2394 "the gRPC control plane task ended before it reported an address; the \
2395 listener on {grpc_bind} never came up"
2396 );
2397 }
2398 };
2399 tracing::info!(
2400 %addr,
2401 control_plane_workers = workers,
2402 control_connection_workers = CONTROL_CONNECTION_WORKERS,
2403 "gunnar grpc control plane listening on its own runtime (a control-plane \
2404 stall cannot reach the git listener, and its calls run on a handler runtime \
2405 that cannot hold the connections)"
2406 );
2407 let mut held = ControlRuntime::new(control_runtime);
2408 held.hold(&mut handler_runtime);
2409 (Some(join), Some(tx), Some((addr, grpc_authenticated)), held)
2410 };
2411
2412 // ── the log level, changeable from here on ─────────────────────────
2413 //
2414 // Started here rather than in `main` because it needs a runtime, and
2415 // deliberately NOT held on `Server`: the task owns two signal streams
2416 // and nothing else, and a level-control task that stopped with the
2417 // server would be one more thing to get wrong at shutdown.
2418 //
2419 // It also has to exist for a NEGATIVE reason. The default disposition
2420 // of both `SIGUSR1` and `SIGUSR2` is to TERMINATE the process, so
2421 // before this an operator who reached for the usual signal killed the
2422 // server they were debugging.
2423 gunnar_common::logging::spawn_signal_control();
2424
2425 // ── the counter heartbeat ──────────────────────────────────────────
2426 let heartbeat = config.metrics_interval.map(|interval| {
2427 let metrics = Arc::clone(&metrics);
2428 // A32-14, stays a task: its handle is kept and aborted at the server's stop.
2429 tokio::spawn(async move { heartbeat(metrics, interval).await })
2430 });
2431
2432 // ── the runner, on the same bus every push publishes on ────────────
2433 //
2434 // Subscribed **before** the listeners bind, so a push that lands in the
2435 // first millisecond is seen. Subscribing after would lose it silently,
2436 // which is the shape of bug that shows up once a month and is blamed on
2437 // the network.
2438 //
2439 // Its own OS thread rather than a tokio task: a build is a blocking
2440 // child process that can run for minutes, and putting it on a runtime
2441 // worker would starve whatever else that runtime is for.
2442 let runner = match config.runner.clone() {
2443 None => None,
2444 Some(runner_config) => {
2445 if let Err(err) = crate::sandbox::available() {
2446 if !runner_config.allow_network {
2447 bail!(
2448 "the runner was asked for hermetic builds and this kernel cannot \
2449 give it a sandbox: {}\n\n\
2450 Refusing to start rather than building with the network up and \
2451 signing records that say otherwise. Pass --runner-allow-network to \
2452 build anyway; every record will say the network was up.",
2453 err.explain()
2454 );
2455 }
2456 }
2457 let stores = Arc::clone(&stores);
2458 let bus = Arc::clone(&bus);
2459 tracing::info!(
2460 workspace = %runner_config.workspace.display(),
2461 builder = %runner_config.builder,
2462 hermetic = !runner_config.allow_network,
2463 "the build runner is watching this server's pushes"
2464 );
2465 // A28 M18, stays a thread: the build runner's watch for the server's life, its handle
2466 // held by the server (`_runner`). A watch that ends is said (`runner::said_end`).
2467 Some(std::thread::spawn(move || {
2468 let end = crate::runner::watch(&stores, &bus, &runner_config, |outcome| {
2469 match outcome {
2470 crate::runner::Watched::Built(built) => tracing::info!(
2471 target: "gunnar::runner",
2472 event = "runner.built",
2473 repo = %built.record.source_repo,
2474 commit = %built.record.source_commit,
2475 recipe = %built.record.recipe,
2476 artifacts = built.record.subjects.len(),
2477 "a push was built and attested"
2478 ),
2479 crate::runner::Watched::Failed { repo, commit, why } => {
2480 // At `error`, and the watch continues: one
2481 // repository's broken recipe must not stop
2482 // every other repository's builds.
2483 tracing::error!(
2484 target: "gunnar::runner",
2485 event = "runner.failed",
2486 %repo, %commit, %why,
2487 "a push could not be built; the runner is still watching"
2488 )
2489 }
2490 crate::runner::Watched::Skipped { repo, r#ref, why } => {
2491 tracing::debug!(
2492 target: "gunnar::runner",
2493 event = "runner.skipped",
2494 %repo, r#ref = %r#ref, %why,
2495 "a ref moved and was not built"
2496 )
2497 }
2498 }
2499 });
2500 match crate::runner::said_end(&end) {
2501 (tracing::Level::ERROR, words) => {
2502 tracing::error!(target: "gunnar::runner", event = "runner.stopped", "{words}")
2503 }
2504 (_, words) => {
2505 tracing::warn!(target: "gunnar::runner", event = "runner.stopped", "{words}")
2506 }
2507 }
2508 }))
2509 }
2510 };
2511 // Held so the thread is not detached from the server's lifetime; the
2512 // watch ends when the bus's last sender goes.
2513 let _runner = runner;
2514
2515 // ── replication, a SUBSCRIBER and nothing else ─────────────────────
2516 //
2517 // Started last, and started on the git runtime deliberately: the work
2518 // it does is a git push over SSH, and the blocking part of it goes to
2519 // the blocking pool. Nothing above this line knows it exists, which is
2520 // the property that makes "replication can never fail a push" true by
2521 // construction rather than by care.
2522 #[cfg(feature = "replication")]
2523 {
2524 use crate::mirror::ReplicationConfig;
2525 let replication_config = match config.replication.clone() {
2526 Some(explicit) => explicit,
2527 None => ReplicationConfig::load(stores.root())
2528 .context("reading the replication configuration")?,
2529 };
2530 // An empty target list leaves the slot empty, which is still replication's
2531 // "a server that replicates nowhere pays nothing": no replicator, no
2532 // subscriber, no task. What it no longer means is that the operator
2533 // has no way to add one.
2534 replication
2535 .apply(replication_config)
2536 .context("starting replication")?;
2537 }
2538
2539 // ── I11's passive heartbeat, if this box was asked for one ─────────
2540 //
2541 // Started AFTER the slot has its configuration, because the task's
2542 // first act is to read the role out of it: a watch spawned before
2543 // `apply` would read the pre-boot file and could leave immediately on a
2544 // box that is about to become a standby.
2545 //
2546 // On the GIT runtime, with the replicator, and for the same reason: the
2547 // probe is a blocking-shaped network wait, and the control plane is the
2548 // surface an operator uses while a box is in trouble.
2549 #[cfg(feature = "replication")]
2550 let standby_watch = config.standby_watch.map(|watch| {
2551 tracing::info!(
2552 target: "gunnar::replication",
2553 primary = %watch.primary,
2554 probe_interval_ms = watch.probe_interval.as_millis() as u64,
2555 promote_after = watch.promote_after,
2556 earliest_takeover_ms = watch.earliest_takeover().as_millis() as u64,
2557 "watching the primary; this box takes over by itself if it stops answering"
2558 );
2559 let slot = Arc::clone(&replication);
2560 // A32-14, stays a task: its handle is kept and aborted at the server's stop.
2561 tokio::spawn(async move { crate::standby_watch::watch(slot, watch).await })
2562 });
2563
2564 // ── 5.4: the boot check, then the pull channel ─────────────────────
2565 //
2566 // After the control plane listens: the boot check reads this box's own
2567 // `Bootstrap.Status`, and a recharge into this box goes through it.
2568 #[cfg(feature = "replication")]
2569 let repair = match (config.repair, grpc_endpoint) {
2570 (Some(cfg), Some((bound, _))) => {
2571 let own = crate::repair::own_addr(bound);
2572 let mut entitlements = crate::repair::Entitlements::of(Arc::clone(&stores));
2573 if let Some(control) = control_for_repair.clone() {
2574 entitlements = entitlements.with_control(control);
2575 }
2576 #[cfg(feature = "metered")]
2577 let mut entitlements = entitlements.with_slot(entitlement.clone());
2578 if let Some(grow) = grow_for_repair.clone() {
2579 entitlements =
2580 entitlements.with_grow(crate::paid_growth::PaidGrowth::new(grow));
2581 }
2582 // The environment's relay; else (a SOLD estate's main, P1) the one the shop FED it,
2583 // read at each mail — not ready until the feed names it, so a resend is refused
2584 // by name before anything is minted.
2585 #[cfg(feature = "fleet-pull")]
2586 let postman: Option<Arc<dyn crate::postman::Postman>> =
2587 match crate::postman::Smtp::from_env() {
2588 Ok(Some(smtp)) => Some(Arc::new(smtp)),
2589 // MAIL-RELAY: the relay the ONE box fed this main again goes over the lay's.
2590 Ok(None) => {
2591 let root = stores.root().to_path_buf();
2592 Some(Arc::new(crate::postman::FedRelay {
2593 fed: move || {
2594 crate::mail_relay::over(crate::feed::Fed::current(), &root)
2595 },
2596 }))
2597 }
2598 Err(why) => {
2599 tracing::warn!(target: crate::repair::TARGET, "no mail relay: a sign_in_again order is refused by name ({why})");
2600 None
2601 }
2602 };
2603 let entitlements = entitlements.with_twin(cfg.twin.clone());
2604 #[cfg(feature = "fleet-pull")]
2605 let entitlements = entitlements.with_postman(postman);
2606 let stores = Arc::clone(&stores);
2607 crate::repair::arm(cfg.clone(), own.clone(), tokio::runtime::Handle::current());
2608 // 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.
2609 Some(tokio::spawn(async move {
2610 crate::repair::boot(Arc::clone(&stores), cfg.clone(), own.clone()).await;
2611 // AIRGAP B5 + M12: only a `fleet-pull` build carries the pull channel (and
2612 // the shared pump it runs on); any other build refuses a named one by name
2613 // (`repair::resolve_fleet`) and pulls nothing.
2614 #[cfg(not(feature = "fleet-pull"))]
2615 let _ = (entitlements, own);
2616 #[cfg(feature = "fleet-pull")]
2617 if let Some(fleet) =
2618 crate::repair::fleet_when_fed(cfg.fleet.clone(), crate::repair::FED_EVERY)
2619 .await
2620 {
2621 let estate = crate::repair::ServerEstate::new(stores.root(), &cfg, own.clone())
2622 .map_err(|why| tracing::warn!(target: crate::repair::TARGET, "the pull channel runs without a repair key: every order is refused by name ({why})"))
2623 .ok();
2624 let (_keep, stop) = tokio::sync::watch::channel(false);
2625 crate::repair::pull_loop(stores, fleet, estate, entitlements, own, stop)
2626 .await;
2627 }
2628 }))
2629 }
2630 (Some(_), None) => {
2631 tracing::warn!(target: "gunnar::repair", "the boot check and the pull channel need the control plane, which is off");
2632 None
2633 }
2634 (None, _) => None,
2635 };
2636
2637 // Outbound webhooks: load webhooks.json and start delivering. An empty file
2638 // (or none) leaves the slot empty — no subscriber, no task — the "a
2639 // server with no webhooks pays nothing" property; a malformed file is a
2640 // hard error here, so a typo is caught at boot and not per delivery.
2641 #[cfg(feature = "webhooks")]
2642 {
2643 let webhook_config = crate::webhook::WebhookConfig::load(stores.root())
2644 .context("reading the webhook configuration")?;
2645 webhook
2646 .apply(webhook_config)
2647 .context("starting outbound webhook delivery")?;
2648 }
2649
2650 // ── publish the counters, so something finally READS them ──────────
2651 //
2652 // One closure, two readers: `config.replication_view` is the console
2653 // band's and `control_view` is `Admin.Stats`'s. Before this the
2654 // counters existed and nothing looked at them, which is how every
2655 // configured standby receiving NOTHING went unnoticed for weeks.
2656 //
2657 // The closure reads the slot on EVERY call rather than closing over one
2658 // `Arc<Replicator>`. That is what keeps the band and `Admin.Stats`
2659 // honest after a live `SetReplication`: a captured replicator would go
2660 // on reporting the standby that was just removed, and report nothing at
2661 // all about the one that was just added. With no target configured the
2662 // slot is empty and the answer is `ReplicationReport::default()` —
2663 // `configured: false`, which is a fact rather than an absence.
2664 #[cfg(feature = "replication")]
2665 for view in [Some(config.replication_view.clone()), control_view.clone()]
2666 .into_iter()
2667 .flatten()
2668 {
2669 let slot = Arc::clone(&replication);
2670 view.fill(move || {
2671 slot.replicator()
2672 .as_deref()
2673 .map(|replicator| {
2674 // The cheap report, plus the ref number PEEKED from the
2675 // last on-demand walk — never a walk of its own. This
2676 // closure runs on every `Admin.Stats` and every band
2677 // frame, so a reader that walked would be the cold-status
2678 // "re-open every store every 60 s from a polling
2679 // caller". `ref_coverage_cached` opens nothing; the
2680 // number is whatever the most recent `--refs` warmed,
2681 // and zeros until the first one.
2682 let mut report = crate::replication_view::report(replicator);
2683 report.refs = replicator.ref_coverage_cached();
2684 report
2685 })
2686 .unwrap_or_default()
2687 });
2688 }
2689 let _ = &control_view;
2690
2691 Ok(Server {
2692 ssh: Some(ssh),
2693 grpc,
2694 grpc_shutdown,
2695 grpc_endpoint,
2696 control_runtime,
2697 stores,
2698 bootstrap_window_open,
2699 host_key_fingerprint,
2700 metrics,
2701 heartbeat,
2702 #[cfg(feature = "replication")]
2703 standby_watch,
2704 #[cfg(feature = "replication")]
2705 repair,
2706 bus,
2707 #[cfg(feature = "http")]
2708 http,
2709 scrape,
2710 #[cfg(feature = "replication")]
2711 replication,
2712 #[cfg(feature = "webhooks")]
2713 webhook,
2714 })
2715 }
2716
2717 /// The event bus.
2718 ///
2719 /// The same object `Events.Watch` streams from, exposed here so an
2720 /// in-process caller (the replicator, and the tests that assert what it
2721 /// publishes) can subscribe without a gRPC listener and a credential in
2722 /// between.
2723 pub fn bus(&self) -> &Arc<EventBus> {
2724 &self.bus
2725 }
2726
2727 /// The replicator, when this server has any standby configured.
2728 ///
2729 /// Exposed so an operator surface, and `tests/replication.rs`, can read the
2730 /// counters and the per-standby status without reaching into the tasks.
2731 #[cfg(feature = "replication")]
2732 pub fn replication(&self) -> Option<Arc<crate::mirror::Replicator>> {
2733 self.replication.replicator()
2734 }
2735
2736 /// The live slot, for a caller that RECONFIGURES rather than only observes:
2737 /// `Mirror.SetReplication`'s in-process half and `tests/replication.rs`.
2738 #[cfg(feature = "replication")]
2739 pub fn replication_slot(&self) -> &Arc<crate::mirror::ReplicationSlot> {
2740 &self.replication
2741 }
2742
2743 /// The live webhook slot, for a caller that reconfigures outbound delivery:
2744 /// `Webhooks.AddWebhook`'s in-process half and `tests/webhooks.rs`.
2745 #[cfg(feature = "webhooks")]
2746 pub fn webhook_slot(&self) -> &Arc<crate::webhook::WebhookSlot> {
2747 &self.webhook
2748 }
2749
2750 /// **The SSH host key this server serves**, `SHA256:…` — the fingerprint a
2751 /// client pins, read off the key that is actually loaded rather than off a
2752 /// file somebody remembers.
2753 pub fn host_key_fingerprint(&self) -> &str {
2754 &self.host_key_fingerprint
2755 }
2756
2757 /// The address the SSH listener actually bound.
2758 pub fn ssh_addr(&self) -> SocketAddr {
2759 self.ssh().local_addr()
2760 }
2761
2762 /// The address the scrape endpoint bound, if it was started.
2763 ///
2764 /// `None` is the ordinary answer: nothing binds unless `--metrics-bind`
2765 /// asked for it. A test that asked for port 0 learns its port here.
2766 pub fn metrics_addr(&self) -> Option<SocketAddr> {
2767 self.scrape.as_ref().map(|(_, addr)| *addr)
2768 }
2769
2770 /// The address the smart-HTTP listener bound, if it was started.
2771 #[cfg(feature = "http")]
2772 pub fn http_addr(&self) -> Option<SocketAddr> {
2773 self.http.as_ref().map(|(_, addr, _)| *addr)
2774 }
2775
2776 /// Whether the smart-HTTP listener terminates TLS itself.
2777 #[cfg(feature = "http")]
2778 pub fn http_tls(&self) -> bool {
2779 self.http.as_ref().is_some_and(|(_, _, tls)| *tls)
2780 }
2781
2782 /// The SSH listener. Present for the whole life of a `Server`; the `Option`
2783 /// exists only so [`Server::shutdown`] can move it out past the [`Drop`]
2784 /// impl.
2785 fn ssh(&self) -> &BoundServer {
2786 self.ssh
2787 .as_ref()
2788 .expect("the SSH listener is taken only by shutdown, which consumes the Server")
2789 }
2790
2791 /// The address the gRPC listener actually bound, if it was started.
2792 pub fn grpc_addr(&self) -> Option<SocketAddr> {
2793 self.grpc_endpoint.map(|(addr, _)| addr)
2794 }
2795
2796 /// The bound gRPC address together with whether that listener
2797 /// authenticates, which is what `serve` prints.
2798 ///
2799 /// One accessor rather than two, because the banner used to carry a
2800 /// hardcoded `UNAUTHENTICATED in v1` beside a control plane that
2801 /// authenticates. A caller that cannot get the address without the verdict
2802 /// cannot print a stale one.
2803 pub fn grpc_endpoint(&self) -> Option<(SocketAddr, bool)> {
2804 self.grpc_endpoint
2805 }
2806
2807 /// Was the unauthenticated enrolment window open when this server started?
2808 ///
2809 /// What `serve` prints beside the gRPC address. `false` is the ordinary
2810 /// answer, which is why the banner says nothing at all in that case.
2811 pub fn bootstrap_window_open(&self) -> bool {
2812 self.bootstrap_window_open
2813 }
2814
2815 /// The stores behind both listeners.
2816 pub fn stores(&self) -> &Arc<ServerStores> {
2817 &self.stores
2818 }
2819
2820 /// Serve until a stop is requested: a signal, or on the appliance the ACPI
2821 /// power button ([`appliance_pid1::runtime::stop_request`]).
2822 ///
2823 /// **Not `ctrl_c()`.** That is `SIGINT` alone, and nothing in an appliance's
2824 /// life sends `SIGINT`: `podman stop` sends `SIGTERM` and every service
2825 /// manager sends `SIGTERM`. A cloud's soft stop (UpCloud `stop`, QEMU
2826 /// `system_powerdown`) is not a signal at all: it is an ACPI power-button
2827 /// press the kernel reports on the `acpi_event` generic-netlink group
2828 /// (corrected 2026-09-28; this said it became `SIGTERM`, and a soft stop
2829 /// then waited out the cloud's timeout and was hard-stopped). PID 1 also
2830 /// gets **no default disposition** for a signal it has not handled — the
2831 /// kernel discards it. See edda's `appliance_pid1::net`.
2832 pub async fn wait(self) -> Result<()> {
2833 let signal = appliance_pid1::runtime::stop_request().await;
2834 tracing::info!(signal, "shutting both listeners down");
2835 self.shutdown().await
2836 }
2837
2838 /// Ask both listeners to stop, and wait for them.
2839 ///
2840 /// # Why the gRPC wait is bounded (measured 2026-08-03)
2841 ///
2842 /// `serve_with_incoming_shutdown` is *graceful*: it stops accepting and
2843 /// then waits for every in-flight request to finish. `Events.Watch` (the event stream)
2844 /// is server-streaming and by design **never finishes**; it ends when the
2845 /// subscriber goes away. So one connected UI made this wait forever, and
2846 /// `gunnar serve` could not be stopped with Ctrl-C at all. Measured: a test
2847 /// binary wedged for eleven minutes with the watch connection still
2848 /// `ESTAB`, no CPU, no progress.
2849 ///
2850 /// Graceful is still tried first — an in-flight `Repos.Create` should be
2851 /// allowed to land — but after [`GRPC_SHUTDOWN_GRACE`] the task is aborted
2852 /// and the process is free to exit. A watcher losing its stream at shutdown
2853 /// is what a watcher is built to handle. A server that cannot be stopped
2854 /// is not.
2855 pub async fn shutdown(mut self) -> Result<()> {
2856 // One last line before the listeners go, so the final numbers are in
2857 // the log rather than only in a process that is about to exit.
2858 if let Some(beat) = self.heartbeat.take() {
2859 beat.abort();
2860 }
2861 // I11's watchdog goes with the listeners. A watch that outlived the
2862 // server it belongs to could promote a box that is shutting down —
2863 // writing `role: primary` to disk as its last act, so the NEXT boot
2864 // comes up writable. Aborting it here is what keeps a shutdown a
2865 // shutdown.
2866 #[cfg(feature = "replication")]
2867 if let Some(watch) = self.standby_watch.take() {
2868 watch.abort();
2869 }
2870 #[cfg(feature = "replication")]
2871 if let Some(repair) = self.repair.take() {
2872 repair.abort();
2873 }
2874 // Replication stops with the server. An in-flight attempt is abandoned
2875 // rather than waited for: the job it was draining is still in the
2876 // queue's shape ("converge this repository") and the next start
2877 // re-derives it from the repository itself.
2878 #[cfg(feature = "replication")]
2879 self.replication.stop();
2880 #[cfg(feature = "webhooks")]
2881 self.webhook.stop();
2882 self.log_metrics();
2883 if let Some(tx) = self.grpc_shutdown.take() {
2884 let _ = tx.send(());
2885 }
2886 if let Some(join) = self.grpc.take() {
2887 let abort = join.abort_handle();
2888 match tokio::time::timeout(GRPC_SHUTDOWN_GRACE, join).await {
2889 Ok(Ok(result)) => result?,
2890 Ok(Err(err)) if err.is_cancelled() => {}
2891 Ok(Err(err)) => return Err(anyhow::anyhow!("the gRPC task panicked: {err}")),
2892 Err(_elapsed) => {
2893 tracing::warn!(
2894 grace = ?GRPC_SHUTDOWN_GRACE,
2895 "the gRPC control plane did not stop gracefully; almost certainly an \
2896 open Events.Watch stream, which by design never ends. Aborting it so \
2897 the process can exit."
2898 );
2899 abort.abort();
2900 }
2901 }
2902 }
2903 // `shutdown_background`, not a drop: `shutdown` is an `async fn` and may
2904 // well be running on a runtime thread, where dropping a `Runtime`
2905 // panics. The listener is already stopped or aborted above, so there is
2906 // nothing left worth blocking for. The `ControlRuntime` guard's own
2907 // `Drop` would do this too; doing it explicitly here just tears the
2908 // control plane down before the SSH listener rather than after.
2909 self.control_runtime.shutdown();
2910 if let Some(ssh) = self.ssh.take() {
2911 ssh.shutdown("gunnar is shutting down")
2912 .await
2913 .context("stopping the SSH listener")?;
2914 }
2915 Ok(())
2916 }
2917
2918 /// What the SSH admission gate has done since start-up.
2919 ///
2920 /// Here so an operator surface (and `tests/`) can read the saturation
2921 /// counters without reaching into `gunnar-ssh`.
2922 pub fn admission(&self) -> gunnar_ssh::AdmissionStats {
2923 self.ssh().admission()
2924 }
2925
2926 /// Every counter this server keeps, read now.
2927 ///
2928 /// One reading of one set: the SSH listener and the git transport write to
2929 /// the same block, so `connections_refused` and `pushes_rejected` are facts
2930 /// about the same server rather than two crates' private tallies.
2931 pub fn metrics(&self) -> MetricsSnapshot {
2932 self.metrics.snapshot()
2933 }
2934
2935 /// The counter block itself, for a caller that wants to keep a handle to it.
2936 pub fn metrics_handle(&self) -> &Arc<ServerMetrics> {
2937 &self.metrics
2938 }
2939
2940 /// Write the counter line now, without waiting for the next heartbeat.
2941 pub fn log_metrics(&self) {
2942 emit_metrics(&self.metrics.snapshot());
2943 }
2944}
2945
2946impl Drop for Server {
2947 fn drop(&mut self) {
2948 if let Some(beat) = self.heartbeat.take() {
2949 beat.abort();
2950 }
2951 // Same reason as in `shutdown`: a watchdog that survives its server
2952 // could promote a box on its way out.
2953 #[cfg(feature = "replication")]
2954 if let Some(watch) = self.standby_watch.take() {
2955 watch.abort();
2956 }
2957 #[cfg(feature = "replication")]
2958 if let Some(repair) = self.repair.take() {
2959 repair.abort();
2960 }
2961 #[cfg(feature = "replication")]
2962 self.replication.stop();
2963 #[cfg(feature = "webhooks")]
2964 self.webhook.stop();
2965 // A `Server` dropped without `shutdown` (a test that panicked, say)
2966 // would otherwise take its control runtime down by a blocking `Drop`,
2967 // which panics on a runtime thread and would bury the real failure. The
2968 // `ControlRuntime` field's own `Drop` already routes through
2969 // `shutdown_background`; this is that same call, made explicit.
2970 self.control_runtime.shutdown();
2971 }
2972}
2973
2974/// **Say, at every start, whether the unauthenticated enrolment door is open.**
2975///
2976/// Silence here is the disease this repository keeps finding. A server standing
2977/// with an open door is a server anybody who can reach the control port can
2978/// take; an operator must never have to guess which state their box is in, and
2979/// on an unenrolled server there is by definition no credential with which to
2980/// ask `Admin.Health`.
2981///
2982/// Split out of [`Server::start`] and given both readings rather than taking
2983/// them, so every branch can be driven from a test on any box. A warning that
2984/// reads its own host can only ever be seen in the state that host happens to
2985/// be in.
2986///
2987/// Returns the line it emitted, so a test asserts on the **text an operator
2988/// sees** rather than on a boolean that proves nothing about the log.
2989pub fn announce_readiness(
2990 readiness: &Readiness,
2991 data_root: &std::path::Path,
2992 grpc_bind: SocketAddr,
2993 no_grpc: bool,
2994) -> String {
2995 match readiness {
2996 Readiness::Ready => {
2997 let line = format!("identity table ready at {}", data_root.display());
2998 tracing::info!(data_root = %data_root.display(), "{line}");
2999 line
3000 }
3001 Readiness::AwaitingBootstrap if no_grpc => {
3002 let line = format!(
3003 "NO IDENTITY HAS EVER BEEN ENROLLED at {}, and the gRPC control plane is \
3004 DISABLED, so there is no way to enrol one. Nobody can reach this server. Start \
3005 without --no-grpc and call Bootstrap.Enroll, or pass --admin-key.",
3006 data_root.display()
3007 );
3008 tracing::error!(data_root = %data_root.display(), "{line}");
3009 line
3010 }
3011 Readiness::AwaitingBootstrap => {
3012 let line = format!(
3013 "NO IDENTITY HAS EVER BEEN ENROLLED at {}. Bootstrap.Enroll on {grpc_bind} is \
3014 open to ANY caller that can reach it, with NO credential, and it creates a \
3015 gunnar-admin. It closes permanently on the first success and deleting \
3016 identities does not reopen it. Enrol now, or bind the control plane where \
3017 nobody else can reach it.",
3018 data_root.display()
3019 );
3020 tracing::warn!(
3021 data_root = %data_root.display(),
3022 %grpc_bind,
3023 bootstrap_window = "OPEN",
3024 "{line}"
3025 );
3026 line
3027 }
3028 Readiness::SealedAndEmpty { record } => {
3029 let who = match record {
3030 Some(record) => format!(
3031 "was bootstrapped by \"{}\" at {}",
3032 record.principal_id, record.at
3033 ),
3034 None => "holds principals but no keys and carries no bootstrap record, so it \
3035 was written before that record existed, and"
3036 .to_owned(),
3037 };
3038 let line = format!(
3039 "this server {who} its identity table at {} now holds no keys. NOBODY CAN \
3040 AUTHENTICATE against it, and the enrolment window stays shut — deleting \
3041 identities does not reopen it. Add a key with `gunnar key add` while the \
3042 server is down.",
3043 data_root.display()
3044 );
3045 tracing::error!(
3046 data_root = %data_root.display(),
3047 bootstrap_window = "SHUT",
3048 "{line}"
3049 );
3050 line
3051 }
3052 }
3053}
3054
3055/// **Turn `--grpc-bind` into the address the listener binds.**
3056///
3057/// The one caller of [`BindSpec::resolve_with`] in the server, and the one
3058/// place `appliance_pid1::net::interface_ipv4` is wired to it. An address resolves to itself
3059/// and never touches netlink; an interface is asked, once, and a refusal stops
3060/// the start rather than widening the bind — the whole point of the spelling.
3061/// Returns the address to bind, or `None` when the named interface exists and
3062/// holds no IPv4 — see [`BindSpec::resolve_with`] for why that is `Ok` and not
3063/// an error, and what the caller owes it.
3064/// **Bind the control plane's listener — on a DEVICE when the spelling named
3065/// one.** `eth1:50051` resolves to the address eth1 holds, and an address alone
3066/// is not the interface: MEASURED 2026-09-05 on the KVM bench rig (two slirp
3067/// NICs, both 10.0.2.15), the listener bound to eth1's address answered on
3068/// eth0's forward and refused eth1's, because the kernel routes by address and
3069/// both interfaces held it. `SO_BINDTODEVICE` makes the socket the interface's
3070/// — the same call the DHCP client makes for its broadcast socket — so the
3071/// declaration `--grpc-bind eth1:50051` means eth1 on every machine, not just on
3072/// one whose subnets happen to differ. An address spelling (`None`) binds as
3073/// before. The socket is made non-blocking and handed to the control runtime's
3074/// I/O driver by `from_std`, on the runtime that calls this.
3075fn bind_grpc_listener(addr: SocketAddr, device: Option<&str>) -> Result<tokio::net::TcpListener> {
3076 use socket2::{Domain, Protocol, Socket, Type};
3077 let domain = if addr.is_ipv6() {
3078 Domain::IPV6
3079 } else {
3080 Domain::IPV4
3081 };
3082 let sock = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
3083 .context("opening the gRPC listener socket")?;
3084 sock.set_reuse_address(true)
3085 .context("SO_REUSEADDR on the gRPC listener")?;
3086 if let Some(dev) = device {
3087 sock.bind_device(Some(dev.as_bytes()))
3088 .with_context(|| format!("SO_BINDTODEVICE {dev} on the gRPC listener"))?;
3089 }
3090 sock.bind(&addr.into())
3091 .with_context(|| format!("bind {addr}"))?;
3092 sock.listen(1024).context("listen on the gRPC listener")?;
3093 sock.set_nonblocking(true)
3094 .context("non-blocking gRPC listener")?;
3095 tokio::net::TcpListener::from_std(sock.into())
3096 .context("registering the gRPC listener with the runtime")
3097}
3098
3099/// How long `serve` waits for the gRPC bind address to turn up on an interface
3100/// before it gives up. Thirty seconds covers a tailnet or WireGuard interface
3101/// that comes up after this service on a cold boot; anything slower is a box
3102/// with a real problem, and the bounded exit hands it to the supervisor.
3103pub const GRPC_BIND_PATIENCE: std::time::Duration = std::time::Duration::from_secs(30);
3104
3105/// The pause between two bind attempts while waiting for the address.
3106pub const GRPC_BIND_RETRY_EVERY: std::time::Duration = std::time::Duration::from_millis(500);
3107
3108/// `EADDRNOTAVAIL` and nothing else. That errno means "no interface on this
3109/// box has that address", which on a cold boot means *yet*. `EADDRINUSE`, a
3110/// bad device, a permission refusal: none of those get better by waiting.
3111fn bind_failed_because_the_address_is_not_here_yet(err: &anyhow::Error) -> bool {
3112 err.chain()
3113 .filter_map(|cause| cause.downcast_ref::<std::io::Error>())
3114 .any(|io| io.kind() == std::io::ErrorKind::AddrNotAvailable)
3115}
3116
3117/// Run `attempt` until it binds, retrying only while the failure is the
3118/// address-not-here-yet errno, every `every`, for at most `patience`.
3119///
3120/// Returns the final result AND the number of retries taken, so a caller (and
3121/// the test) can tell a first-try bind from a waited-for one.
3122async fn bind_when_the_address_arrives<T, F>(
3123 mut attempt: F,
3124 patience: std::time::Duration,
3125 every: std::time::Duration,
3126) -> (Result<T>, u32)
3127where
3128 F: FnMut() -> Result<T>,
3129{
3130 let started = std::time::Instant::now();
3131 let mut retries = 0u32;
3132 loop {
3133 match attempt() {
3134 Ok(bound) => {
3135 if retries > 0 {
3136 tracing::info!(
3137 retries,
3138 waited_ms = started.elapsed().as_millis() as u64,
3139 "the gRPC bind address arrived; the control plane is bound"
3140 );
3141 }
3142 return (Ok(bound), retries);
3143 }
3144 Err(err) if bind_failed_because_the_address_is_not_here_yet(&err) => {
3145 if started.elapsed() >= patience {
3146 return (
3147 Err(err.context(format!(
3148 "the address was on no interface for {} s ({retries} retries)",
3149 patience.as_secs()
3150 ))),
3151 retries,
3152 );
3153 }
3154 if retries == 0 {
3155 tracing::warn!(
3156 patience_s = patience.as_secs(),
3157 "the gRPC bind address is on no interface yet (EADDRNOTAVAIL); waiting for \
3158 it. A tailnet or VPN interface that comes up after this service is the \
3159 usual cause"
3160 );
3161 }
3162 retries += 1;
3163 tokio::time::sleep(every).await;
3164 }
3165 Err(err) => return (Err(err), retries),
3166 }
3167 }
3168}
3169
3170#[cfg(test)]
3171mod grpc_bind_patience_tests {
3172 use super::*;
3173 use std::time::Duration;
3174
3175 fn not_here_yet() -> anyhow::Error {
3176 anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrNotAvailable))
3177 .context("bind 100.64.0.1:50061")
3178 }
3179
3180 /// The address arrives on the third try: bound, and the retry count says
3181 /// the wait was real (a wrapper that never retried would return 0 here).
3182 #[test]
3183 fn an_address_that_arrives_later_is_bound_and_the_retries_are_counted() {
3184 let rt = tokio::runtime::Builder::new_current_thread()
3185 .enable_all()
3186 .build()
3187 .unwrap();
3188 let mut left = 2;
3189 let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
3190 || {
3191 if left > 0 {
3192 left -= 1;
3193 Err(not_here_yet())
3194 } else {
3195 Ok(7u8)
3196 }
3197 },
3198 Duration::from_secs(5),
3199 Duration::from_millis(1),
3200 ));
3201 assert_eq!(bound.unwrap(), 7);
3202 assert_eq!(retries, 2);
3203 }
3204
3205 /// Any other errno fails on the first try, with zero retries: waiting on
3206 /// `EADDRINUSE` would hide a second server on the port for thirty seconds.
3207 #[test]
3208 fn an_address_in_use_is_not_waited_for() {
3209 let rt = tokio::runtime::Builder::new_current_thread()
3210 .enable_all()
3211 .build()
3212 .unwrap();
3213 let mut calls = 0;
3214 let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
3215 || -> Result<u8> {
3216 calls += 1;
3217 Err(anyhow::Error::from(std::io::Error::from(
3218 std::io::ErrorKind::AddrInUse,
3219 )))
3220 },
3221 Duration::from_secs(5),
3222 Duration::from_millis(1),
3223 ));
3224 assert!(bound.is_err());
3225 assert_eq!(retries, 0);
3226 assert_eq!(calls, 1);
3227 }
3228
3229 /// The patience is a bound: an address that never comes gives up with the
3230 /// wait named in the error, not a hang.
3231 #[test]
3232 fn an_address_that_never_comes_gives_up_after_the_patience() {
3233 let rt = tokio::runtime::Builder::new_current_thread()
3234 .enable_all()
3235 .build()
3236 .unwrap();
3237 let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
3238 || -> Result<u8> { Err(not_here_yet()) },
3239 Duration::from_millis(30),
3240 Duration::from_millis(5),
3241 ));
3242 let msg = format!("{:#}", bound.unwrap_err());
3243 assert!(msg.contains("on no interface for 0 s"), "{msg}");
3244 assert!(retries >= 3, "retries={retries}");
3245 }
3246
3247 /// The real socket path: TEST-NET-1 is on no interface anywhere, and the
3248 /// kernel's answer is the errno the wait is keyed on. This is the red that
3249 /// proves the classifier reads the errno that `bind_grpc_listener` actually
3250 /// produces, not one the test invented.
3251 #[test]
3252 fn the_real_bind_error_for_an_absent_address_is_classified_as_not_here_yet() {
3253 let rt = tokio::runtime::Builder::new_current_thread()
3254 .enable_io()
3255 .build()
3256 .unwrap();
3257 let _guard = rt.enter();
3258 let err = bind_grpc_listener("192.0.2.1:0".parse().unwrap(), None)
3259 .expect_err("TEST-NET-1 is on no interface");
3260 assert!(
3261 bind_failed_because_the_address_is_not_here_yet(&err),
3262 "{err:#}"
3263 );
3264 let in_use = anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrInUse));
3265 assert!(!bind_failed_because_the_address_is_not_here_yet(&in_use));
3266 }
3267}
3268
3269#[cfg(test)]
3270mod grpc_bind_device_tests {
3271 use super::*;
3272
3273 /// A device that does not exist is refused BY NAME before anything binds —
3274 /// the red that proves the device half is wired at all.
3275 #[test]
3276 fn a_device_the_machine_does_not_have_is_refused_by_name() {
3277 let rt = tokio::runtime::Builder::new_current_thread()
3278 .enable_io()
3279 .build()
3280 .unwrap();
3281 let _guard = rt.enter();
3282 let err = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("nosuchdev0"))
3283 .expect_err("a nonexistent device cannot be bound");
3284 let msg = format!("{err:#}");
3285 assert!(msg.contains("SO_BINDTODEVICE nosuchdev0"), "{msg}");
3286 }
3287
3288 /// The address-only spelling binds exactly as before, and on the loopback
3289 /// device by name too (the one device every box has).
3290 #[test]
3291 fn an_address_binds_and_so_does_the_loopback_device() {
3292 let rt = tokio::runtime::Builder::new_current_thread()
3293 .enable_io()
3294 .build()
3295 .unwrap();
3296 let _guard = rt.enter();
3297 let plain = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), None).unwrap();
3298 assert!(plain.local_addr().unwrap().port() != 0);
3299 match bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("lo")) {
3300 Ok(l) => assert!(l.local_addr().unwrap().port() != 0),
3301 // Kernels before 5.7 gate SO_BINDTODEVICE behind CAP_NET_RAW; a
3302 // permission refusal is that gate, not a wiring defect.
3303 Err(e) => assert!(format!("{e:#}").contains("SO_BINDTODEVICE lo"), "{e:#}"),
3304 }
3305 }
3306}
3307
3308fn resolve_grpc_bind(spec: &BindSpec) -> Result<Option<SocketAddr>> {
3309 match spec
3310 .resolve_with(|name| {
3311 appliance_pid1::net::interface_ipv4(name).map_err(|e| format!("{e:#}"))
3312 })
3313 .map_err(|why| anyhow::anyhow!("{why}"))?
3314 {
3315 gunnar_common::bindspec::Resolved::Bound(addr) => Ok(Some(addr)),
3316 gunnar_common::bindspec::Resolved::NoAddress { interface, why } => {
3317 // ERROR and not WARN: an operator who asked for a control plane
3318 // has not got one, and nothing else on the box will say so.
3319 tracing::error!(
3320 interface = %interface,
3321 "{why} The gRPC control plane will NOT start. The git listener is unaffected \
3322 and this server still serves and authenticates pushes and fetches over SSH."
3323 );
3324 Ok(None)
3325 }
3326 }
3327}
3328
3329/// `--admin-key` is either the key line itself or `@<path>`.
3330fn read_key_argument(spec: &str) -> Result<String> {
3331 match spec.strip_prefix('@') {
3332 Some(path) => std::fs::read_to_string(path)
3333 .with_context(|| format!("reading the public key from {path}"))
3334 .map(|text| text.trim().to_owned()),
3335 None => Ok(spec.trim().to_owned()),
3336 }
3337}
3338
3339/// **The startup banner's host-key line.**
3340///
3341/// The fingerprint IN FULL, never shortened: the dashboard's `short_fp` is for
3342/// recognising a key you already know, and this line exists to be the first
3343/// place anybody learns it. It is read off a VNC screenshot by a human and
3344/// typed into a `known_hosts` line, and twelve characters cannot be typed into
3345/// anything.
3346///
3347/// The word `sshkey` leads it so a screenshot can be searched for one token.
3348///
3349/// Why this is worth a line of banner at all: `known_hosts` records no
3350/// provenance. Once written, a key taken off the wire by `ssh-keyscan` and a
3351/// key read off a console are the same bytes, for ever after, and nothing in
3352/// the file can tell the two apart. The only thing that makes a forge push
3353/// mean anything is WHERE the fingerprint came from — a record written by the
3354/// install (`deploy/upcloud`'s receipt) or a reading off the box's own console.
3355/// Never from whatever answered on `:2222`.
3356pub(crate) fn host_key_banner_line(fingerprint: &str) -> String {
3357 format!("sshkey {fingerprint} (pin THIS; a key scanned off the port proves nothing)")
3358}
3359
3360/// The host key, as russh's `ssh-key` sees it.
3361fn russh_private_key(openssh: &[u8]) -> Result<gunnar_ssh::russh::keys::PrivateKey> {
3362 gunnar_ssh::russh::keys::PrivateKey::from_openssh(openssh)
3363 .map_err(|err| anyhow::anyhow!("the stored host key did not parse: {err}"))
3364}
3365
3366/// Run `gunnar serve` to completion.
3367pub async fn run(stores: Arc<ServerStores>, config: ServeConfig, check_only: bool) -> Result<()> {
3368 // The descriptor limit, raised once. `ServerStores::open` already did this
3369 // (its handle cache is sized from the answer); calling it here is the
3370 // guarantee that the `limits.nofile` line — a WARN row on the dashboard
3371 // when the limit is low — exists before a listener binds, whichever path
3372 // built `stores`.
3373 let _ = crate::limits::nofile();
3374 // ── THE NARRATOR, BEFORE ANYTHING ELSE ──────────────────────────────────
3375 //
3376 // MEASURED 2026-09-23: `boot::ship_boot_narration` ends at `end of
3377 // hand-off` — the moment the DHCP lease makes the wire usable — and the
3378 // estate's twin died in the silence after it, having read its config,
3379 // bound its listeners, taken its role, held its certificate and started
3380 // following. None of that was said. From here on every boundary is.
3381 //
3382 // The sink is `korp.log_udp=` off the kernel command line and nothing
3383 // else. Absent ⇒ nothing is armed, no thread is started, no datagram is
3384 // sent, and the two file readers (the ESP's BOOTLOG.TXT and /data) carry
3385 // the story as they always did. That is the off switch.
3386 use crate::narrate::Stage;
3387 crate::narrate::install();
3388 crate::narrate::data_home(stores.root());
3389 crate::narrate::at(
3390 Stage::Config,
3391 &format!(
3392 "data root {}, engine {:?}, ssh {}, grpc {:?}, twin_role={}, is_twin={}",
3393 stores.root().display(),
3394 stores.engine(),
3395 config.ssh_bind,
3396 config.grpc_bind,
3397 config.twin_role,
3398 stores.is_twin(),
3399 ),
3400 );
3401 // ── PID 1's own duties, before anything binds ───────────────────────────
3402 //
3403 // The network first and by hand, because the interface has to exist before
3404 // a listener can be reached over it and on this appliance nothing else
3405 // brings it up: measured on the reference desktop 2026-08-03, the kernel is built with
3406 // `CONFIG_IP_PNP` off, so `ip=` on the command line configures nothing.
3407 // A refused static plan or malformed GUNNAR_NET is fatal on purpose: the
3408 // alternative is a server that starts, logs "ssh 0.0.0.0:2222" and can
3409 // never be reached, which is the same failure `Readiness::RefuseToServe`
3410 // exists to prevent. A DHCP timeout is NOT fatal — bring_up_network warns
3411 // and serves unreachable, the missing-GUNNAR_NET posture — because there
3412 // the world declined to answer; the declaration did not lie.
3413 crate::narrate::at(
3414 Stage::Network,
3415 "bringing the declared interfaces up over netlink",
3416 );
3417 let dhcp_session = crate::boot::bring_up_network()?;
3418 // The lease. The UDP narration wire is retired (2026-09-24): the installer
3419 // collects this box's whole boot log over its own mutual TLS.
3420 crate::narrate::at(
3421 Stage::Lease,
3422 &format!("{} DHCP lease(s) held", dhcp_session.len()),
3423 );
3424 crate::narrate::at(
3425 Stage::Sink,
3426 "the boot log is kept for the installer (its bootstrap listener serves it)",
3427 );
3428 // **The clock, and here rather than anywhere later.** It is after the
3429 // network because it is fetched over the network, and before every listener
3430 // because `gunnar_grpc::auth::MAX_CLOCK_SKEW_MS` is ±5 minutes: an appliance
3431 // that serves a single request on a clock two hours out refuses a genuine
3432 // credential and a genuine Stripe webhook signature, which is precisely what
3433 // gunnar.rs measured on 2026-09-14. Never fatal, never a `?`: see
3434 // `crate::clock`'s note on why a failed clock fetch must not cost a forge.
3435 crate::clock::synchronise_at_boot().await;
3436 // And again when a credential is refused for skew, and every six hours: a
3437 // boot-time step fixes the clock the box boots with, never the drift of the
3438 // months until the next re-image. Background threads; see `crate::clock`.
3439 crate::clock::arm_triggers();
3440 // **The bind spelling becomes an address here, and not one line earlier.**
3441 // `--grpc-bind eth1:50051` names an interface whose address arrives with
3442 // the DHCP lease above; asking before `bring_up_network` would find an
3443 // interface with no address every time. Asking AFTER is also what makes
3444 // the refusal actionable: if it fires, the lease genuinely did not land,
3445 // and the warning `bring_up_network` already logged says why.
3446 let mut config = config;
3447 // **A bind by ROLE** (`utility:50051`, the lean estate, A19 row 17) becomes this box's
3448 // utility NIC here, after the network is up: `eth1` on a box with two NICs, `eth0` on a
3449 // box laid behind its front with its utility NIC alone.
3450 config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)
3451 .context("resolving the --grpc-bind role to this box's NIC")?;
3452 if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
3453 tracing::info!(interface = %interface, "the control plane's utility door is {interface} on this box");
3454 }
3455 // **An interface whose lease PID 1 still asks for** (its boot DHCP failed — estate-box3
3456 // D4, a box rebooted while its network was down) is WAITED for, here on the appliance's
3457 // path as in `Server::start`: the control plane binds once the address arrives.
3458 if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
3459 let has_address = |i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
3460 if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
3461 tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
3462 while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
3463 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
3464 }
3465 }
3466 }
3467 let resolved = resolve_grpc_bind(&config.grpc_bind).with_context(|| {
3468 "resolving --grpc-bind after the network came up. A medium that names a NIC this \
3469 machine does not have is a declaration that lies, and gunnar stops rather than \
3470 falling back to a wildcard bind: an unauthenticated control plane on every \
3471 interface is the defect the interface spelling exists to close."
3472 })?;
3473 let resolved_grpc_bind = match resolved {
3474 Some(addr) => addr,
3475 None => {
3476 config.no_grpc = true;
3477 gunnar_grpc::DEFAULT_BIND
3478 }
3479 };
3480 // Pin the answer into the config so `Server::start`, the dashboard and the
3481 // readiness line all report the address the listener took rather than the
3482 // spelling it was asked for — and so the netlink question is asked once.
3483 if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
3484 config.grpc_bind_device = Some(interface.clone());
3485 }
3486 config.grpc_bind = BindSpec::Addr(resolved_grpc_bind);
3487 if !config.no_grpc && !resolved_grpc_bind.ip().is_loopback() {
3488 tracing::info!(
3489 grpc_bind = %resolved_grpc_bind,
3490 "gRPC control plane will bind a NON-loopback address"
3491 );
3492 }
3493 // Whether /data survived the last install (the ESP record korp-installer
3494 // left) — one event, one dashboard row, before the band starts.
3495 crate::boot::announce_install_mode(stores.root());
3496 // The start-up disk test, appliance only (same gate as the dashboard, or
3497 // `GUNNAR_DISKTEST=1`): one thread, one `disktest` row, never waited for.
3498 // A32-14, stays a thread: the one-shot start-up disk test, appliance only; it writes its row and ends, never waited for.
3499 crate::disktest::spawn(stores.root().to_path_buf());
3500 if appliance_pid1::net::is_init() {
3501 tracing::info!(
3502 "running as pid 1: no init system underneath, so this process handles SIGTERM and \
3503 reaps re-parented orphans itself"
3504 );
3505 }
3506 // Installed unconditionally. A reaper that only exists at pid 1 is a reaper
3507 // no test on a developer machine ever runs.
3508 // A32-14, stays a task: the pid 1 reaper for the whole process; its handle is kept and aborted at the stop.
3509 let reaper = tokio::spawn(appliance_pid1::runtime::reaper());
3510 // **The soft-delete sweep**: the half of `repo delete` that gives the bytes
3511 // back. Hourly, and once now. Without it a retention window would be a
3512 // promise to keep repositories forever, and `/data` would fill with
3513 // repositories nobody can list.
3514 let tombstone_sweep = crate::stores::spawn_tombstone_sweep(Arc::clone(&stores));
3515 // **The flow flush**: the half of the activity ledger that survives a
3516 // restart. The counters a transfer bumps live in atomics in this process;
3517 // this copies them into the control store every minute, and rolls the UTC
3518 // day on a box that is idle at midnight. See `stores::spawn_flow_flush`.
3519 // The band's fold, refreshed by the flush task once a minute. `None` until
3520 // the first flush: a row of zeroes before anything has been folded would
3521 // claim the box served nobody today.
3522 let flow_band: Arc<std::sync::Mutex<Option<gunnar_core::Fleet>>> =
3523 Arc::new(std::sync::Mutex::new(None));
3524 let flow_flush = crate::stores::spawn_flow_flush(Arc::clone(&stores), Arc::clone(&flow_band));
3525 config
3526 .flow_view
3527 .fill(move || *flow_band.lock().expect("the flow band"));
3528 // Kept out of the move into `Server::start` below, for the shutdown flush.
3529 let flow_at_exit = Arc::clone(stores.flow());
3530 // The VGA status screen, appliance only (PID 1 booted by korp-start) or
3531 // `GUNNAR_DASHBOARD=<device>`. One thread, fed by the log tap; the lease
3532 // line above is already in the tap's buffer and is its first row. Never
3533 // fatal, never waited for — see `crate::dashboard`.
3534 // The band starts before the server does, so it is handed a VIEW and the
3535 // replicator is dropped into it below, once `Server::start` has one. A box
3536 // with no standby leaves it unfilled and the band draws no `standby` row.
3537 let replication_view = config.replication_view.clone();
3538 // A32-14, stays a thread: the VGA status screen, appliance only, fed by the log tap for the process's life; never fatal.
3539 crate::dashboard::spawn(crate::dashboard::Facts {
3540 ssh_bind: config.ssh_bind,
3541 grpc_bind: (!config.no_grpc).then_some(resolved_grpc_bind),
3542 data_root: stores.root().to_path_buf(),
3543 build: gunnar_common::BUILD_REF.to_owned(),
3544 engine: Some(stores.engine()),
3545 replication: replication_view.clone(),
3546 tenants: config.tenants_view.clone(),
3547 flow: config.flow_view.clone(),
3548 });
3549 // Exactly when the address is DHCP-learned, a task keeps the lease alive
3550 // (RFC 2131 §4.4.5 — appliance_pid1::runtime::dhcp_renewal). Static and unconfigured paths
3551 // return an EMPTY list above and gain no task: renewal exists only where a
3552 // lease does — and a two-NIC appliance holds TWO leases, each with its own
3553 // T1, so there is one task per lease rather than one task guessing which
3554 // interface it is keeping alive.
3555 #[cfg(feature = "dhcp-client")]
3556 let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
3557 .into_iter()
3558 // A32-14, stays a task: one lease's renewal for the process's life; the handles are kept and aborted at the stop.
3559 .map(|session| tokio::spawn(crate::boot::dhcp_renewal(session)))
3560 .collect();
3561 // Built without `dhcp-client`: `DhcpSession` is UNINHABITED, so this is the
3562 // compiler proving there is no lease rather than a runtime check claiming
3563 // it. The `Some` arm is `match session {}` — unreachable by construction —
3564 // and the abort sites below need no cfg of their own.
3565 #[cfg(not(feature = "dhcp-client"))]
3566 let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
3567 .into_iter()
3568 .map(|session| match session {})
3569 .collect();
3570
3571 // ── the tunnel, AFTER the outer address ─────────────────────────────────
3572 //
3573 // Order is load-bearing and not alphabetical: a WireGuard peer is reached
3574 // over the outer address, so a handshake attempted before DHCP has
3575 // finished has no source address to leave from. It is strictly additive —
3576 // no `wg0.conf` in the data root means one INFO line and nothing else
3577 // changes — and it is deliberately NOT fatal when the kernel refuses,
3578 // because an appliance that will not serve git because a tunnel failed is
3579 // worse than one that serves on its outer address and says so loudly.
3580 //
3581 // Nothing below has to be told about it: both listeners bind `0.0.0.0`
3582 // (`ServeConfig::default`, ssh 0.0.0.0:2222), so `wg0` is served the
3583 // moment it has an address.
3584 // **This binding is the tunnel's lifetime.** The data plane is a thread in
3585 // THIS process now, not a device the kernel owns: `Tunnel` holds the pump
3586 // and the tun fd, and dropping the last handle stops the pump and deletes
3587 // `wg0`. The clone below goes to a watcher task that RETURNS after the
3588 // first handshake, so if this binding were consumed instead of kept the
3589 // tunnel would come up, announce itself and vanish a few seconds later.
3590 // It stays in scope for the whole of `run`, which is the process.
3591 // A pairing box's `wg0.conf` is written by the pairing loop once the
3592 // installer's door has assigned the address (pairing.rs); on a reboot it
3593 // is already here and this lane raises it like any other.
3594 #[cfg(feature = "wg-client")]
3595 let tunnel = crate::wg::bring_up(stores.root())?;
3596 #[cfg(feature = "wg-client")]
3597 let handshake = tunnel.clone().map(|t| {
3598 // A32-14, stays a task: the tunnel's first-handshake watcher; it returns after the handshake, its handle is aborted at the stop.
3599 tokio::spawn(crate::wg::handshake_watch(
3600 t,
3601 20,
3602 std::time::Duration::from_secs(3),
3603 ))
3604 });
3605 // Built without `wg-client`: there is no tunnel type, no boringtun and no
3606 // `wg_appliance_core` in this binary at all, so there is nothing to bring
3607 // up, nothing to watch and no banner line to write. No `tunnel` binding
3608 // exists in this build — only the watcher handle the two abort paths below
3609 // reach for, and it is permanently `None`.
3610 #[cfg(not(feature = "wg-client"))]
3611 let handshake: Option<tokio::task::JoinHandle<()>> = None;
3612
3613 // Read off both BEFORE the move below, so the role line can be narrated
3614 // after the listeners are up: `Server::start` consumes `stores` and
3615 // `config`, and the role is a property of the boot rather than of the
3616 // server object.
3617 let config_twin_role = config.twin_role;
3618 let stores_is_twin = stores.is_twin();
3619 // `Server::start` fills `config.replication_view` — the same object the
3620 // band above was handed — once the replicator exists.
3621 let server = Server::start(stores, config).await?;
3622 // The staleness alarm: one WARN into the log and one event-stream event the first
3623 // time a standby stops taking bytes — with "last successful replication:
3624 // NEVER" as its own message. Harmless when nothing is configured: the view
3625 // reports no standbys and the task never fires.
3626 let staleness_watch = crate::replication_view::spawn_staleness_watch(
3627 replication_view.clone(),
3628 Arc::clone(server.bus()),
3629 );
3630 // The banner is BUILT once and written twice (one implementation, one writer): to stdout,
3631 // which on the appliance is `/dev/console` = the serial line that
3632 // `installtest/src/iso.rs` reads its boot proof off, and — appliance only —
3633 // mirrored to `/dev/tty0` so a KVM GRAPHICS console ends on gunnar's own
3634 // status instead of the kernel's last printk. Measured 2026-08-12: with the
3635 // banner on serial alone, a VGA console showed a healthy boot log that never
3636 // once said "gunnar".
3637 let mut banner = String::new();
3638 {
3639 use std::fmt::Write as _;
3640 writeln!(banner, "ssh {}", server.ssh_addr()).ok();
3641 // **THE HOST KEY, ON A CHANNEL PRODUCTION HAS.** Directly under the
3642 // address it belongs to, because the two together are the whole of
3643 // what a client needs to trust this box.
3644 //
3645 // MEASURED 2026-09-23. The fingerprint was logged and nowhere else:
3646 // `tracing` → stderr → `/dev/console` → the LAST `console=` on the
3647 // command line → the serial line, and the UpCloud shape has no serial
3648 // line (a panic there prints PID 1's first `print!` to nothing; the
3649 // ISO's own gate VMs are the only boxes that ever had it, in their
3650 // `serial.log`). So in production the one fact a client must pin
3651 // reached NOBODY, and this laptop's `known_hosts` has eight
3652 // `ssh-keyscan`-shaped lines for `[git.gunnar.rs]:2222` to prove what
3653 // fills the gap.
3654 //
3655 // This banner is the channel that answers: it is printed on stdout AND
3656 // mirrored to `/dev/tty0` (`appliance_pid1::console::mirror_to_gfx_console`, a few
3657 // lines below), which is the VNC screen of an UpCloud VM. The
3658 // WireGuard public key is already here for exactly this argument — on
3659 // a shell-less appliance the banner is the only channel that can
3660 // answer "is the key on this box the one I enrolled?".
3661 //
3662 // The narration wire (`korp.log_udp=`) was considered FIRST and is
3663 // not usable: it is armed and flushed long before this point, but the
3664 // one door onto it (`appliance_pid1::narration::narration_door`) classifies any
3665 // line carrying `sha256:` as `pubkey` and replaces it with a withheld
3666 // notice, the collector refuses it identically
3667 // (`<ops>/xtask/src/bootnarr.rs::secret_class`), and the 40-character
3668 // opaque-run rule catches the fingerprint's body with the label
3669 // stripped off — so no spelling escapes. That door is an OWNER RULE
3670 // ("a PUBLIC verifying key … is withheld anyway, because it is not
3671 // progress"), spelled in three repositories. Putting the fingerprint
3672 // on that wire needs the rule reversed and all three vocabularies
3673 // changed together; it is not something this line may do quietly.
3674 writeln!(
3675 banner,
3676 "{}",
3677 host_key_banner_line(server.host_key_fingerprint())
3678 )
3679 .ok();
3680 match server.grpc_endpoint() {
3681 // The open enrolment window comes FIRST on this line, before the
3682 // reassuring "authenticated": a server that has never been enrolled is
3683 // one anybody who can reach the port can take, and an operator reading
3684 // three lines of banner must not have to know that "authenticated"
3685 // does not cover `Bootstrap.Enroll`.
3686 Some((addr, _)) if server.bootstrap_window_open() => writeln!(
3687 banner,
3688 "grpc {addr} (NOT ENROLLED — Bootstrap.Enroll is open to anyone who can reach \
3689 this port. `gunnar control bootstrap <name>` claims it, once.)"
3690 )
3691 .ok(),
3692 Some((addr, true)) => writeln!(
3693 banner,
3694 "grpc {addr} (authenticated: sshsig + the ACL model)"
3695 )
3696 .ok(),
3697 Some((addr, false)) => writeln!(
3698 banner,
3699 "grpc {addr} (UNAUTHENTICATED — the escape hatch is on)"
3700 )
3701 .ok(),
3702 None => writeln!(banner, "grpc disabled").ok(),
3703 };
3704 // Only when it is on, like metrics below. Without this line a server asked
3705 // to bind HTTP on port 0 has bound a listener nobody can find: ssh, grpc,
3706 // metrics and the log level were all announced, and the one listener a
3707 // stock `git` client talks to was not.
3708 #[cfg(feature = "http")]
3709 if let Some(addr) = server.http_addr() {
3710 if server.http_tls() {
3711 writeln!(
3712 banner,
3713 "https {addr} (TLS terminated here; clients pin the issuer)"
3714 )
3715 .ok();
3716 } else {
3717 writeln!(banner, "http {addr}").ok();
3718 }
3719 }
3720 // Only when it is on: a line reading "metrics disabled" on every
3721 // server that never asked for the feature is banner an operator learns to
3722 // skip, and the banner's whole value is that it is short enough to read.
3723 if let Some(addr) = server.metrics_addr() {
3724 writeln!(
3725 banner,
3726 "metrics http://{addr}{} (no authentication{})",
3727 crate::scrape::SCRAPE_PATH,
3728 if addr.ip().is_loopback() {
3729 ""
3730 } else {
3731 "; NOT on loopback"
3732 }
3733 )
3734 .ok();
3735 }
3736 // The tunnel, and ONLY when there is one — same rule as metrics above.
3737 //
3738 // It carries the appliance's WireGuard PUBLIC KEY, which is not
3739 // decoration: peering is mutual allowlisting by public key, and a key
3740 // the hub does not have in its `[Peer]` stanza gets **silence** rather
3741 // than a rejection. So "is the key on this box the one I enrolled?" is
3742 // the first question of every failed tunnel, and on a shell-less
3743 // appliance the banner — serial AND the mirrored graphics console — is
3744 // the only channel that can answer it.
3745 #[cfg(feature = "wg-client")]
3746 if let Some(tunnel) = &tunnel {
3747 writeln!(banner, "{}", tunnel.banner_line()).ok();
3748 }
3749 // On stdout beside the addresses rather than in the log: this is
3750 // the one fact an operator needs BEFORE the incident, and by the time they
3751 // want it the log is the thing they cannot read enough of. It also warns
3752 // them off the trap — both signals default to terminating a process, and
3753 // that is what they did to this one until today.
3754 writeln!(
3755 banner,
3756 "log {} (SIGUSR1 → one rung more verbose, SIGUSR2 → back; pid {})",
3757 gunnar_common::logging::current().unwrap_or_else(|| "?".to_owned()),
3758 std::process::id()
3759 )
3760 .ok();
3761 }
3762 print!("{banner}");
3763 appliance_pid1::console::mirror_to_gfx_console(&banner);
3764 // **ONE LINE PER LISTENER, WITH THE ADDRESS IT TOOK.** Not the spelling it
3765 // was asked for: `--grpc-bind eth1:50051` and `10.13.9.24:50051` are two
3766 // different facts and only the second one can be dialled. A listener the
3767 // build does not carry says nothing rather than saying `disabled`, so a
3768 // reader counting listeners counts the ones that exist.
3769 crate::narrate::at(Stage::Listener, &format!("ssh BOUND {}", server.ssh_addr()));
3770 match server.grpc_endpoint() {
3771 Some((addr, _)) => crate::narrate::at(
3772 Stage::Listener,
3773 &format!(
3774 "grpc BOUND {addr}{}",
3775 if server.bootstrap_window_open() {
3776 " (NOT ENROLLED — Bootstrap.Enroll is open)"
3777 } else {
3778 ""
3779 }
3780 ),
3781 ),
3782 None => crate::narrate::at(Stage::Listener, "grpc NOT bound — the control plane is off"),
3783 }
3784 #[cfg(feature = "http")]
3785 if let Some(addr) = server.http_addr() {
3786 crate::narrate::at(
3787 Stage::Listener,
3788 &format!(
3789 "{} BOUND {addr}",
3790 if server.http_tls() { "https" } else { "http" }
3791 ),
3792 );
3793 }
3794 if let Some(addr) = server.metrics_addr() {
3795 crate::narrate::at(Stage::Listener, &format!("metrics BOUND {addr}"));
3796 }
3797 // **THE ROLE**, from the two places that decide it: what the medium
3798 // declared and what the data home already is. They are narrated together
3799 // because a disagreement between them is the interesting case — a box
3800 // installed from main's medium onto a disk that already holds a twin's
3801 // certificate is exactly the shape the estate re-images into.
3802 crate::narrate::at(
3803 Stage::Role,
3804 &format!(
3805 "{} (declared twin_role={}, data home says is_twin={})",
3806 if stores_is_twin || config_twin_role {
3807 "TWIN"
3808 } else {
3809 "PRIMARY"
3810 },
3811 config_twin_role,
3812 stores_is_twin,
3813 ),
3814 );
3815 if check_only {
3816 reaper.abort();
3817 tombstone_sweep.abort();
3818 flow_flush.abort();
3819 staleness_watch.abort();
3820 for renewal in &renewals {
3821 renewal.abort();
3822 }
3823 if let Some(handshake) = &handshake {
3824 handshake.abort();
3825 }
3826 server.shutdown().await?;
3827 return Ok(());
3828 }
3829 // **SERVING.** The end of the boot story and the beginning of the ticks:
3830 // from here the only thing the box is expected to say is one short line a
3831 // minute, and the ear's threshold is measured against that. A box that
3832 // reached here and then went quiet is a DEATH AT A NAMED PLACE — which is
3833 // the reading the estate did not have on 2026-09-23.
3834 crate::narrate::at(
3835 Stage::Serving,
3836 &format!("up at {} s", crate::narrate::uptime_s(),),
3837 );
3838 let result = server.wait().await;
3839 reaper.abort();
3840 tombstone_sweep.abort();
3841 // One last flush before the process goes. The flush COPIES rather than
3842 // adds, so doing it here as well as on the tick cannot double a count, and
3843 // it is what stops an orderly shutdown losing the last minute of today.
3844 if let Err(err) = flow_at_exit.flush(gunnar_core::now_ms()) {
3845 tracing::warn!(
3846 target: "gunnar::store",
3847 event = "flow.flush_failed",
3848 %err,
3849 "the flow ledger could not be written on shutdown; up to one flush interval of \
3850 today's activity counters is lost"
3851 );
3852 }
3853 flow_flush.abort();
3854 staleness_watch.abort();
3855 for renewal in &renewals {
3856 renewal.abort();
3857 }
3858 if let Some(handshake) = &handshake {
3859 handshake.abort();
3860 }
3861 result
3862}
3863
3864/// The public key argument reader, exposed for the wiring tests.
3865#[cfg(test)]
3866mod tests {
3867 use super::read_key_argument;
3868 use super::*;
3869
3870 /// **The banner line an operator reads off the VNC screen.**
3871 ///
3872 /// The fingerprint must be there IN FULL and unmangled: this line is the
3873 /// only production channel that carries it (the log goes to a serial
3874 /// console the UpCloud shape does not have, and the narration wire's one
3875 /// door withholds anything containing `sha256:`), and a fingerprint that
3876 /// has been shortened for display cannot be pinned by anybody.
3877 ///
3878 /// Red-first: shortening the fingerprint the way the dashboard's
3879 /// `short_fp` does, or dropping it from the line, fails here.
3880 #[test]
3881 fn the_banner_names_the_host_key_in_full() {
3882 const FP: &str = "SHA256:z/fSv0Z0RZS+Lccbc6ZoOWwt/fbj1VFJGSBKQbo4icE";
3883 let line = host_key_banner_line(FP);
3884 assert!(line.contains(FP), "{line}");
3885 assert!(line.starts_with("sshkey "), "one searchable token: {line}");
3886 assert!(
3887 !line.contains('\n'),
3888 "one line, or the banner is not a banner: {line}"
3889 );
3890 // And it warns off the thing that fills known_hosts when this line is
3891 // missing, because a fingerprint with no provenance is not evidence.
3892 assert!(
3893 line.contains("scanned off the port proves nothing"),
3894 "{line}"
3895 );
3896 }
3897
3898 /// The visible limit has to sit **inside** the pool that carries
3899 /// it, or the invisible ceiling is back, at whichever of the two numbers is
3900 /// smaller, which is the one nobody is looking at.
3901 #[test]
3902 fn the_blocking_pool_is_larger_than_the_transfer_limit() {
3903 let transfers = gunnar_ssh::SshConfig::default()
3904 .max_concurrent_transfers
3905 .expect("the appliance default must have a transfer limit");
3906 assert!(
3907 transfers < GIT_BLOCKING_THREADS,
3908 "max_concurrent_transfers ({transfers}) is not below GIT_BLOCKING_THREADS \
3909 ({GIT_BLOCKING_THREADS}), so the blocking pool becomes the real ceiling again and \
3910 the admission counters would show a server that is never full while every transfer \
3911 queues"
3912 );
3913 assert!(
3914 GIT_BLOCKING_THREADS >= transfers * 2,
3915 "the blocking pool ({GIT_BLOCKING_THREADS}) leaves no headroom over \
3916 {transfers} concurrent transfers for the other blocking work a transfer needs: \
3917 pack indexing, ref transactions, the arrow log"
3918 );
3919 }
3920
3921 /// The control plane must get a pool that is bounded on both sides — and
3922 /// at least FOUR workers (R5.1-G): two slow synchronous calls (a snapshot,
3923 /// a disk walk) held both workers of the old floor, and every other call —
3924 /// the console's page loads, monetize's `Entitlement.Set` — waited behind
3925 /// them.
3926 #[test]
3927 fn the_control_plane_gets_its_own_bounded_pool() {
3928 let workers = control_plane_workers();
3929 assert!(
3930 (4..=8).contains(&workers),
3931 "control_plane_workers() returned {workers}; below four, two slow synchronous calls \
3932 hold every worker and the whole administrative surface queues behind them, and an \
3933 unbounded pool recreates the starved listener with extra steps"
3934 );
3935 }
3936
3937 #[test]
3938 fn a_literal_key_and_an_at_path_read_the_same() {
3939 let dir = tempfile::tempdir().expect("a temp dir");
3940 let path = dir.path().join("id.pub");
3941 std::fs::write(&path, "ssh-ed25519 AAAAC3Nz literal\n").expect("write");
3942 assert_eq!(
3943 read_key_argument("ssh-ed25519 AAAAC3Nz literal").expect("literal"),
3944 read_key_argument(&format!("@{}", path.display())).expect("from file"),
3945 "`--admin-key @file` and `--admin-key '<line>'` must produce the same key, or a \
3946 bootstrap that works one way fails the other"
3947 );
3948 }
3949
3950 /// The 32-CPUs-of-affinity-on-8-CPUs-of-quota condition, seen **both** ways.
3951 ///
3952 /// A warning that reads its own host can only ever be observed in the state
3953 /// that host is in, and a build box that happens to be unconstrained would
3954 /// have exercised the silent branch for ever while the loud one rotted.
3955 #[test]
3956 fn a_quota_far_narrower_than_the_affinity_is_reported_and_a_matching_one_is_not() {
3957 // The measured bad case: `podman run --cpus 8` on a 32-core host.
3958 assert_eq!(
3959 spread_budget(Some(32), Some(8.0)),
3960 Some((32, 8)),
3961 "eight CPUs of quota spread over thirty-two is the condition that cost 2.7x the \
3962 CPU per request at 32 clients; it must be said out loud"
3963 );
3964 // The same budget as a cpuset: nothing to report.
3965 assert_eq!(spread_budget(Some(8), Some(8.0)), None);
3966 // Exactly at the line, and one either side of it, because `>=` and `>`
3967 // are one character apart and both compile.
3968 assert_eq!(spread_budget(Some(16), Some(8.0)), Some((16, 8)));
3969 assert_eq!(spread_budget(Some(15), Some(8.0)), None);
3970 // A fractional quota rounds UP to the CPUs an operator would name.
3971 assert_eq!(spread_budget(Some(32), Some(1.5)), Some((32, 2)));
3972 // No quota at all is the ordinary case and is not a finding. Neither is
3973 // an unreadable affinity: half a comparison would be a number invented
3974 // to fill the other half.
3975 assert_eq!(spread_budget(Some(32), None), None);
3976 assert_eq!(spread_budget(None, Some(8.0)), None);
3977 }
3978
3979 /// `Cpus_allowed_list` is ranges *and* singletons, and the two mixed.
3980 ///
3981 /// Against literals rather than against this host: the reference desktop reports a plain
3982 /// `0-31`, and a parser that counted ranges and silently dropped singletons
3983 /// passed on this box while answering 0 for a server pinned to one CPU.
3984 #[test]
3985 fn a_cpu_list_is_counted_whether_it_is_ranges_singletons_or_both() {
3986 assert_eq!(count_cpu_list("0-31"), Some(32));
3987 assert_eq!(count_cpu_list("0-7"), Some(8));
3988 assert_eq!(
3989 count_cpu_list("3"),
3990 Some(1),
3991 "one pinned CPU is one CPU, not none"
3992 );
3993 assert_eq!(count_cpu_list("0,2,4,6"), Some(4));
3994 assert_eq!(count_cpu_list("0-3,8,10-11"), Some(7));
3995 assert_eq!(
3996 count_cpu_list(" 0-3 , 8 \n"),
3997 Some(5),
3998 "the kernel line is padded"
3999 );
4000 // Unreadable is None, never a number: a guessed count is the one thing
4001 // that turns this whole check into a lie.
4002 assert_eq!(count_cpu_list(""), None);
4003 assert_eq!(count_cpu_list("banana"), None);
4004 assert_eq!(
4005 count_cpu_list("7-3"),
4006 None,
4007 "a reversed range is not 5 CPUs"
4008 );
4009 }
4010
4011 #[test]
4012 fn the_affinity_of_this_process_is_read_as_a_count_of_cpus() {
4013 let n = cpus_allowed().expect("Linux always publishes Cpus_allowed_list");
4014 assert!(
4015 (1..=1 << 20).contains(&n),
4016 "this process reports {n} allowed CPUs, which is not a count of CPUs"
4017 );
4018 // The whole comparison rests on this being the *affinity* and not the
4019 // machine, so it must never exceed what the machine has.
4020 let online = std::thread::available_parallelism()
4021 .map(|n| n.get())
4022 .unwrap_or(1);
4023 assert!(
4024 n >= online,
4025 "the affinity ({n}) is narrower than what std reports as available ({online}), \
4026 which would make every comparison against it too quiet"
4027 );
4028 }
4029
4030 #[test]
4031 fn a_missing_key_file_is_an_error_not_an_empty_key() {
4032 let err = read_key_argument("@/nonexistent/gunnar/key.pub")
4033 .expect_err("a missing key file must fail");
4034 // An empty string here would reach `bootstrap_admin` and fail there
4035 // with a parse error, but it would also be indistinguishable from "no
4036 // key was asked for" in any caller that checked for emptiness.
4037 assert!(format!("{err:#}").contains("reading the public key"));
4038 }
4039}
4040
4041/// **The front doors' gate chain, in the one order it is ever built** (one implementation):
4042/// the disk-headroom gate first — every build — then, on a `metered` build
4043/// with a monetize key, the entitlement gate. SSH and smart-HTTP take the same
4044/// `Control`, so a push is refused for the same reason on both; and
4045/// `tests/entitlement.rs` builds the chain through this function rather than
4046/// by hand, so what it asserts is the wiring `serve` runs, not a copy of it.
4047pub fn front_control(
4048 stores: &crate::stores::ServerStores,
4049 anonymous: gunnar_ssh::AnonymousPolicy,
4050 disk: Arc<gunnar_ssh::DiskHeadroomGate>,
4051 #[cfg(feature = "metered")] entitlement: Option<&Arc<crate::entitlement::EntitlementSlot>>,
4052) -> gunnar_ssh::Control {
4053 #[allow(unused_mut)]
4054 // **The data set's space ledger** (cov-13): the disk gate's own, so the door
4055 // before the first byte and every receive loop after it count one set of
4056 // promises against one floor. Installed here because this is the one place
4057 // the chain is built.
4058 stores.install_space_ledger(Arc::clone(disk.ledger()));
4059 let control = stores
4060 .ssh_control()
4061 .with_anonymous_policy(anonymous)
4062 .with_plan_gate(disk as Arc<dyn gunnar_ssh::PlanGate>);
4063 #[cfg(feature = "metered")]
4064 let control = match entitlement {
4065 Some(slot) => control.with_plan_gate(Arc::clone(slot) as Arc<dyn gunnar_ssh::PlanGate>),
4066 None => control,
4067 };
4068 // **And the same chain for a mirror's worker** (2026-10-02): a mirror sync
4069 // grows a namespace exactly as a push does, so it is held to the same
4070 // room by the same gates — `mirror_fetch::room_for` reads what is
4071 // installed here. One chain, built in this one place.
4072 stores.install_plan_gates(control.plan_gates().to_vec());
4073 control
4074}
4075
4076/// **The replication door's address, written where `twin certify` reads it**:
4077/// the bound port always; the HOSTS only while a second door is open on the
4078/// tunnel — then `<tunnel>,<utility>`, the twin's order of preference (it
4079/// dials the first that answers and falls back to the next). With no tunnel
4080/// door nothing is recorded and the twin dials the host it was joined to.
4081pub fn record_replication_door(
4082 root: &std::path::Path,
4083 utility: SocketAddr,
4084 tunnel: Option<SocketAddr>,
4085) -> Result<()> {
4086 crate::twin_tls::record_replication_port(root, Some(utility.port()))?;
4087 let hosts: Vec<std::net::IpAddr> = match tunnel {
4088 Some(t) => [t.ip(), utility.ip()]
4089 .into_iter()
4090 .filter(|ip| !ip.is_unspecified())
4091 .collect(),
4092 None => vec![],
4093 };
4094 crate::twin_tls::record_replication_host(root, &hosts)
4095}
4096
4097/// **The mutual replication listener** (owner 2026-09-22, amended 2026-09-24):
4098/// serves on `listener` — the UTILITY address — until the box stops, and opens
4099/// a SECOND door on the tunnel address while there is one
4100/// ([`crate::tunnel_addr::replication_extra`]), closing it when the tunnel
4101/// goes away or moves. The utility door is never taken away: raising wg0
4102/// happens before any handshake, and a tunnel that never answers must not
4103/// cost the twin the path that works.
4104#[allow(clippy::too_many_arguments)]
4105pub async fn replication_door(
4106 listener: tokio::net::TcpListener,
4107 base: SocketAddr,
4108 explicit: bool,
4109 wg_only: bool,
4110 ctx: gunnar_grpc::ServiceContext,
4111 request_ids: gunnar_grpc::RequestIdSource,
4112 trust: gunnar_grpc::Trust,
4113 root: std::path::PathBuf,
4114 stop: tokio::sync::watch::Receiver<bool>,
4115) {
4116 // The door's certificate is the twin's PAIRING: one a principal no longer
4117 // holds (a nuked twin's, superseded when its replacement was certified —
4118 // twin reborn 2026-09-30) is refused here even when a header rides along.
4119 let mut ctx = ctx;
4120 ctx.config = ctx
4121 .config
4122 .clone()
4123 .with_registered_client_certificate_required(true);
4124 let serve = move |listener: tokio::net::TcpListener,
4125 mut stop: tokio::sync::watch::Receiver<bool>| {
4126 let router = gunnar_grpc::router_with_trust(ctx.clone(), request_ids, Some(&trust));
4127 async move {
4128 let router = match router {
4129 Ok(r) => r,
4130 Err(err) => {
4131 tracing::error!(%err, "the mutual replication listener could not be configured; no twin can follow this box");
4132 return;
4133 }
4134 };
4135 let at = listener
4136 .local_addr()
4137 .map(|a| a.to_string())
4138 .unwrap_or_default();
4139 let served = router
4140 .serve_with_incoming_shutdown(
4141 tokio_stream::wrappers::TcpListenerStream::new(listener),
4142 async move {
4143 let _ = stop.wait_for(|s| *s).await;
4144 },
4145 )
4146 .await;
4147 if let Err(err) = served {
4148 tracing::error!(%err, %at, "a mutual replication door stopped");
4149 }
4150 }
4151 };
4152 replication_doors_wg(
4153 listener,
4154 base,
4155 explicit,
4156 wg_only,
4157 root,
4158 crate::tunnel_addr::subscribe(),
4159 stop,
4160 serve,
4161 )
4162 .await
4163}
4164
4165/// [`replication_door`]'s rule with the serving and the tunnel watch handed in
4166/// — the tests drive it with a plain accept loop and their own tunnel channel.
4167pub async fn replication_doors<S, F>(
4168 listener: tokio::net::TcpListener,
4169 base: SocketAddr,
4170 explicit: bool,
4171 root: std::path::PathBuf,
4172 tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
4173 stop: tokio::sync::watch::Receiver<bool>,
4174 serve: S,
4175) where
4176 S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
4177 F: std::future::Future<Output = ()> + Send + 'static,
4178{
4179 replication_doors_wg(listener, base, explicit, false, root, tunnel, stop, serve).await
4180}
4181
4182/// [`replication_doors`] under a posture: `wg_only` closes the utility door
4183/// and serves the tunnel's alone ([`doors`]); the twin is told the tunnel host
4184/// only.
4185#[allow(clippy::too_many_arguments)]
4186pub async fn replication_doors_wg<S, F>(
4187 listener: tokio::net::TcpListener,
4188 base: SocketAddr,
4189 explicit: bool,
4190 wg_only: bool,
4191 root: std::path::PathBuf,
4192 tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
4193 stop: tokio::sync::watch::Receiver<bool>,
4194 serve: S,
4195) where
4196 S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
4197 F: std::future::Future<Output = ()> + Send + 'static,
4198{
4199 let record = move |utility: SocketAddr, door: Option<SocketAddr>| {
4200 let at = if wg_only {
4201 record_replication_door_wg_only(&root, utility.port(), door)
4202 } else {
4203 record_replication_door(&root, utility, door)
4204 };
4205 if let Err(err) = at {
4206 tracing::error!(%err, "the replication doors changed but their addresses could not be recorded");
4207 }
4208 };
4209 doors(
4210 listener,
4211 base,
4212 wg_only,
4213 tunnel,
4214 stop,
4215 serve,
4216 move |b, t| crate::tunnel_addr::replication_extra(b, explicit, t),
4217 record,
4218 "mutual replication listener",
4219 )
4220 .await
4221}
4222
4223/// **The replication door under `--wg-only`, recorded**: the port, and the
4224/// tunnel's address as the ONE host (none while there is no tunnel: the twin
4225/// dials the host it was joined to, the primary's overlay address).
4226pub fn record_replication_door_wg_only(
4227 root: &std::path::Path,
4228 port: u16,
4229 tunnel: Option<SocketAddr>,
4230) -> Result<()> {
4231 crate::twin_tls::record_replication_port(root, Some(port))?;
4232 let hosts: Vec<std::net::IpAddr> = tunnel.map(|t| t.ip()).into_iter().collect();
4233 crate::twin_tls::record_replication_host(root, &hosts)
4234}
4235
4236/// **The control plane's doors** (`:50051`). Not `wg_only`: the listener as
4237/// bound serves for the process lifetime, as it always has. `wg_only` (owner,
4238/// 2026-10-04): the bound — utility — listener is CLOSED, so a plain-IP
4239/// connection is refused by the kernel, and a door on the tunnel's address,
4240/// same port, serves while there is a tunnel ([`doors`]).
4241pub async fn control_doors<S, F>(
4242 listener: tokio::net::TcpListener,
4243 base: SocketAddr,
4244 wg_only: bool,
4245 tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
4246 stop: tokio::sync::watch::Receiver<bool>,
4247 serve: S,
4248) where
4249 S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
4250 F: std::future::Future<Output = ()> + Send + 'static,
4251{
4252 doors(
4253 listener,
4254 base,
4255 wg_only,
4256 tunnel,
4257 stop,
4258 serve,
4259 |_, _| None,
4260 |_, _| {},
4261 "control plane",
4262 )
4263 .await
4264}
4265
4266/// **One rule for a listener and the tunnel**, shared by the replication
4267/// listener and the control plane.
4268///
4269/// * not `wg_only`: `listener` (the utility address) serves until `stop`;
4270/// `extra` names a SECOND door on the tunnel while there is one (the
4271/// replication rule, [`crate::tunnel_addr::replication_extra`]) or none;
4272/// * `wg_only`: `listener` is CLOSED at once — nothing serves on the utility
4273/// address, and the kernel refuses a plain-IP connection — and a door on the
4274/// tunnel's address, same port, serves while there is a tunnel, moving with
4275/// it.
4276///
4277/// `record` hears every change: `(the utility address, the tunnel door)`.
4278#[allow(clippy::too_many_arguments)]
4279pub async fn doors<S, F, X, R>(
4280 listener: tokio::net::TcpListener,
4281 base: SocketAddr,
4282 wg_only: bool,
4283 mut tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
4284 mut stop: tokio::sync::watch::Receiver<bool>,
4285 serve: S,
4286 extra: X,
4287 mut record: R,
4288 what: &'static str,
4289) where
4290 S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
4291 F: std::future::Future<Output = ()> + Send + 'static,
4292 X: Fn(SocketAddr, Option<std::net::Ipv4Addr>) -> Option<SocketAddr>,
4293 R: FnMut(SocketAddr, Option<SocketAddr>),
4294{
4295 // The port actually bound (a `:0` bind in a test picks one).
4296 let port = listener
4297 .local_addr()
4298 .map(|a| a.port())
4299 .unwrap_or(base.port());
4300 let base = SocketAddr::new(base.ip(), port);
4301 let utility = if wg_only {
4302 drop(listener);
4303 tracing::warn!(%base, "{what}: --wg-only — the utility door is CLOSED; it answers only on the WireGuard tunnel's address");
4304 None
4305 } else {
4306 // A32-14, stays a task: a listener (the utility door); it ends at the server's stop.
4307 Some(tokio::spawn(serve(listener, stop.clone())))
4308 };
4309 // The open tunnel door: its address and the switch that closes it.
4310 let mut door: Option<(SocketAddr, tokio::sync::watch::Sender<bool>)> = None;
4311 let mut first = true;
4312 loop {
4313 let t = *tunnel.borrow_and_update();
4314 let want = if wg_only {
4315 t.map(|t| SocketAddr::new(std::net::IpAddr::V4(t), port))
4316 } else {
4317 extra(base, t)
4318 };
4319 if first || door.as_ref().map(|(a, _)| *a) != want {
4320 first = false;
4321 if let Some((at, close)) = door.take() {
4322 let _ = close.send(true);
4323 if wg_only {
4324 tracing::warn!(%at, "{what}: the tunnel went away or moved: its door is closed, and with --wg-only nothing serves until it is back");
4325 } else {
4326 tracing::warn!(%at, "{what}: the tunnel went away or moved: the door on it is closed; the utility door {base} keeps serving");
4327 }
4328 }
4329 if let Some(at) = want {
4330 match tokio::net::TcpListener::bind(at).await {
4331 Ok(l) => {
4332 let (close, closed) = tokio::sync::watch::channel(false);
4333 // A32-14, stays a task: a listener (the tunnel door); it ends when its close switch is sent.
4334 tokio::spawn(serve(l, closed));
4335 if wg_only {
4336 tracing::info!(%at, "{what}: up on the WireGuard tunnel's address, and only there (--wg-only)");
4337 } else {
4338 tracing::warn!(%at, %base, "{what} ALSO on the tunnel's address (no restart); the utility door stays");
4339 }
4340 door = Some((at, close));
4341 }
4342 Err(err) => {
4343 tracing::error!(%err, %at, "{what}: the tunnel's door could not bind")
4344 }
4345 }
4346 }
4347 record(base, door.as_ref().map(|(a, _)| *a));
4348 }
4349 tokio::select! {
4350 _ = async { let _ = stop.wait_for(|s| *s).await; } => break,
4351 changed = tunnel.changed() => {
4352 if changed.is_err() {
4353 let _ = stop.wait_for(|s| *s).await;
4354 break;
4355 }
4356 }
4357 }
4358 }
4359 if let Some((_, close)) = door.take() {
4360 let _ = close.send(true);
4361 }
4362 if let Some(u) = utility {
4363 let _ = u.await;
4364 }
4365}
4366
4367#[cfg(test)]
4368mod replication_door_tests {
4369 use super::*;
4370 use std::net::Ipv4Addr;
4371 use tokio::io::AsyncWriteExt as _;
4372
4373 /// A door that answers every connection with one byte, until closed.
4374 async fn answering(
4375 listener: tokio::net::TcpListener,
4376 mut stop: tokio::sync::watch::Receiver<bool>,
4377 ) {
4378 loop {
4379 tokio::select! {
4380 _ = async { let _ = stop.wait_for(|s| *s).await; } => return,
4381 got = listener.accept() => if let Ok((mut c, _)) = got { let _ = c.write_all(b"R").await; },
4382 }
4383 }
4384 }
4385
4386 /// Does a door answer at `at`? A closed one refuses at once; the 10 s bounds are hang guards
4387 /// for a door that accepted and never wrote, long enough that a loaded box's slow answer is
4388 /// still an answer (2 s was not, at load 16 under `nice`).
4389 async fn answers(at: SocketAddr) -> bool {
4390 use tokio::io::AsyncReadExt as _;
4391 let Ok(Ok(mut c)) = tokio::time::timeout(
4392 std::time::Duration::from_secs(10),
4393 tokio::net::TcpStream::connect(at),
4394 )
4395 .await
4396 else {
4397 return false;
4398 };
4399 let mut b = [0u8; 1];
4400 matches!(
4401 tokio::time::timeout(std::time::Duration::from_secs(10), c.read_exact(&mut b)).await,
4402 Ok(Ok(_))
4403 )
4404 }
4405
4406 async fn settle() {
4407 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
4408 }
4409
4410 /// **Wait for the doors to act**, not for a fixed time (R5.1, 2026-10-07). The doors act on a
4411 /// change when their task next runs, and on a loaded box (load 32 on dwarf) that was not
4412 /// within the 150 ms `settle` these asserts used to trust: "the tunnel door answers" failed
4413 /// with the door about to open. `cond` is asked every 20 ms until it holds; the two minutes
4414 /// are a hang guard (a test that takes under a second alone took 30 s and more at load 16
4415 /// under `nice`), so only a door that never acts fails.
4416 async fn eventually<F, Fut>(mut cond: F) -> bool
4417 where
4418 F: FnMut() -> Fut,
4419 Fut: std::future::Future<Output = bool>,
4420 {
4421 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(120);
4422 loop {
4423 if cond().await {
4424 return true;
4425 }
4426 if std::time::Instant::now() >= deadline {
4427 return false;
4428 }
4429 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
4430 }
4431 }
4432
4433 /// **WireGuard-only (owner 2026-10-04)**: the utility door is CLOSED — a
4434 /// plain-IP connection to the utility address is refused — and only the
4435 /// tunnel door answers, recorded as the one host the twin is told.
4436 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4437 async fn under_wg_only_the_utility_door_is_closed_and_only_the_tunnel_door_answers() {
4438 let dir = tempfile::tempdir().unwrap();
4439 let root = dir.path().to_path_buf();
4440 let utility = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4441 let base = utility.local_addr().unwrap();
4442 let (tunnel_tx, tunnel_rx) = tokio::sync::watch::channel(None::<Ipv4Addr>);
4443 let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
4444 let doors = tokio::spawn(replication_doors_wg(
4445 utility,
4446 base,
4447 false,
4448 true,
4449 root.clone(),
4450 tunnel_rx,
4451 stop_rx,
4452 answering,
4453 ));
4454 let refused = |at: SocketAddr| async move {
4455 matches!(tokio::net::TcpStream::connect(at).await, Err(e) if e.kind() == std::io::ErrorKind::ConnectionRefused)
4456 };
4457
4458 assert!(
4459 eventually(|| refused(base)).await,
4460 "no tunnel: the utility door is closed, not served"
4461 );
4462 assert_eq!(crate::twin_tls::replication_host(&root), None);
4463
4464 let t = Ipv4Addr::new(127, 0, 0, 2);
4465 tunnel_tx.send_replace(Some(t));
4466 let on_tunnel = SocketAddr::new(t.into(), base.port());
4467 assert!(
4468 eventually(|| answers(on_tunnel)).await,
4469 "the tunnel door answers"
4470 );
4471 assert!(
4472 eventually(|| refused(base)).await,
4473 "and a plain-IP connection is refused"
4474 );
4475 assert!(
4476 eventually(|| std::future::ready(
4477 crate::twin_tls::replication_host(&root).as_deref() == Some("127.0.0.2")
4478 ))
4479 .await,
4480 "the twin is told the tunnel alone: {:?}",
4481 crate::twin_tls::replication_host(&root)
4482 );
4483
4484 // The CONTROL PLANE follows the same rule on its own port.
4485 let control = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4486 let cbase = control.local_addr().unwrap();
4487 let (_ctx, crx) = tokio::sync::watch::channel(Some(t));
4488 let (cstop_tx, cstop_rx) = tokio::sync::watch::channel(false);
4489 let cdoors = tokio::spawn(control_doors(
4490 control, cbase, true, crx, cstop_rx, answering,
4491 ));
4492 assert!(
4493 eventually(|| answers(SocketAddr::new(t.into(), cbase.port()))).await,
4494 "the control plane answers on the tunnel"
4495 );
4496 assert!(
4497 eventually(|| refused(cbase)).await,
4498 "and refuses the plain IP"
4499 );
4500
4501 stop_tx.send_replace(true);
4502 cstop_tx.send_replace(true);
4503 tokio::time::timeout(std::time::Duration::from_secs(5), doors)
4504 .await
4505 .expect("the doors stop")
4506 .unwrap();
4507 tokio::time::timeout(std::time::Duration::from_secs(5), cdoors)
4508 .await
4509 .expect("the control doors stop")
4510 .unwrap();
4511 }
4512
4513 /// Fix 2026-09-24: raising the tunnel no longer MOVES the replication
4514 /// listener off the utility address. With the tunnel down, not bindable, up,
4515 /// and gone again, the utility door answers throughout; the tunnel door is
4516 /// an addition, recorded `<tunnel>,<utility>` only while it is open.
4517 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4518 async fn the_utility_door_answers_whatever_the_tunnel_does() {
4519 let dir = tempfile::tempdir().unwrap();
4520 let root = dir.path().to_path_buf();
4521 let utility = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4522 let base = utility.local_addr().unwrap();
4523 let (tunnel_tx, tunnel_rx) = tokio::sync::watch::channel(None::<Ipv4Addr>);
4524 let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
4525 let doors = tokio::spawn(replication_doors(
4526 utility,
4527 base,
4528 false,
4529 root.clone(),
4530 tunnel_rx,
4531 stop_rx,
4532 answering,
4533 ));
4534
4535 // Tunnel down: the utility door answers, no host recorded.
4536 assert!(eventually(|| answers(base)).await);
4537 assert!(
4538 eventually(|| std::future::ready(
4539 crate::twin_tls::replication_port(&root) == Some(base.port())
4540 ))
4541 .await,
4542 "the twin is told the utility port: {:?}",
4543 crate::twin_tls::replication_port(&root)
4544 );
4545 assert_eq!(crate::twin_tls::replication_host(&root), None);
4546
4547 // A tunnel address this box cannot bind (raised elsewhere / not yet on
4548 // the interface): the utility door still answers.
4549 tunnel_tx.send_replace(Some(Ipv4Addr::new(10, 254, 254, 254)));
4550 settle().await;
4551 assert!(
4552 eventually(|| answers(base)).await,
4553 "a tunnel door that cannot bind must not cost the utility door"
4554 );
4555 assert_eq!(crate::twin_tls::replication_host(&root), None);
4556
4557 // Tunnel up: BOTH answer; the twin is told tunnel first, utility next.
4558 let t = Ipv4Addr::new(127, 0, 0, 2);
4559 tunnel_tx.send_replace(Some(t));
4560 let on_tunnel = SocketAddr::new(t.into(), base.port());
4561 assert!(
4562 eventually(|| answers(on_tunnel)).await,
4563 "the tunnel door answers"
4564 );
4565 assert!(answers(base).await, "and the utility door still does");
4566 assert!(
4567 eventually(|| std::future::ready(
4568 crate::twin_tls::replication_host(&root).as_deref() == Some("127.0.0.2,127.0.0.1")
4569 ))
4570 .await,
4571 "the twin is told tunnel first, utility next: {:?}",
4572 crate::twin_tls::replication_host(&root)
4573 );
4574
4575 // Tunnel gone: its door closes, the utility door stays.
4576 tunnel_tx.send_replace(None);
4577 assert!(
4578 eventually(|| async move { !answers(on_tunnel).await }).await,
4579 "the tunnel door is closed"
4580 );
4581 assert!(answers(base).await);
4582 assert!(
4583 eventually(|| std::future::ready(crate::twin_tls::replication_host(&root).is_none()))
4584 .await,
4585 "no host once the tunnel is gone: {:?}",
4586 crate::twin_tls::replication_host(&root)
4587 );
4588
4589 stop_tx.send_replace(true);
4590 tokio::time::timeout(std::time::Duration::from_secs(5), doors)
4591 .await
4592 .expect("the doors stop")
4593 .unwrap();
4594 }
4595}