Skip to main content

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