nordisk/gunnar
crates/gunnar-server/src/control_plane.rs
//! `gunnar-grpc`'s [`ControlPlane`] seam, over `gunnar-core` and
//! `gunnar-store`: the adapter its module docs asked for.
//!
//! `gunnar-grpc` was written before `gunnar-core` existed and defined a local
//! trait describing what the seven services need. Its own docs say the trait
//! must become *"a thin adapter, one impl in this crate"* and must never be a
//! second source of truth for what a `Repo` is. This is that adapter, living
//! one level up because it is the binary that knows both halves.
//!
//! # What is real here, and what is not
//!
//! | surface | status |
//! |---|---|
//! | namespaces, repos | real, [`RepoCatalog`] |
//! | principals, keys, grants | real, [`AuthTable`] |
//! | `Refs.List` / `Get` / `ReadObject` / `Log` | real, the repository's own stores |
//! | `Admin.Stats` / `StorageInfo` | real, measured off the `fs` backend |
//! | `Refs.Diff` | **`Unimplemented`**: no diff engine on the server side |
//! | `Policy.*` | **`Unimplemented`**: `gunnar-core` has no protection store |
//! | `Mirror.*` | **`Unimplemented`**: mirroring needs `gunnar-client`, which an `S` crate must never link |
//!
//! Every one of those is an explicit [`ControlError::Unimplemented`] and never
//! an empty success. An empty ref list is indistinguishable from a working
//! server with no refs (measured, not asserted), and the same is true of "0 mirrors synced".
//!
//! # Three model mismatches, recorded rather than papered over
//!
//! 1. **`RepoPath` cannot express a nested namespace.** `gunnar-proto`'s
//! `RepoPath::validate` rejects a `/` inside `namespace`, but nesting makes
//! `team/sub/repo.git` legal over SSH. Such a repository is reachable by git
//! and invisible to gRPC. That is a `gunnar-proto` change, not one to make
//! here.
//! 2. **`Scope::Server` has no `RepoPath`.** A server-wide grant is carried as
//! an empty namespace and an empty repo, which `MemoryControlPlane` refuses
//! on the way in. The test double is the one that is wrong: without a
//! server-scoped grant the bootstrap admin cannot create the first namespace
//! (the secret-store seam).
//! 3. **`Namespace` in `gunnar-core` carries only a name** — no description, no
//! creation time — so those two proto fields come back empty rather than
//! invented.
use std::sync::Arc;
use gunnar_core::{Grant, HashKind, Level, PrincipalKind, RepoRef, Scope, UnixMillis, Visibility};
use gunnar_grpc::seam::ControlCounts;
use gunnar_grpc::MergedRef;
use gunnar_grpc::{ControlError, ControlPlane, ControlResult, NewRepo as ProtoNewRepo};
use gunnar_proto::{
BranchProtection, Commit, FileChange, GrantLevel, MirrorConfig, Namespace, Principal,
PublicKey, PullRequest, PullRequestState, ReadLfsObjectResponse, ReadObjectResponse,
RefNamePolicy, RefTarget, Repo, RepoPath, RepoStorage, Signature, StorageInfoResponse,
};
use gunnar_store::git_storage_trait::{GitServe, ObjType, RefRow};
use gunnar_store::object;
use gunnar_wire::RefSnapshot;
use crate::storage::OpenRepo;
use crate::stores::ServerStores;
/// **How long abdication waits for the standby's queues to drain**, seconds.
///
/// A ceiling, not a promise: a standby that is down must not wedge the
/// hand-off, so after this the roster is cleared anyway (a recovered standby
/// replays from this box when it is re-added the other way round). Long enough
/// that a reachable standby with a normal backlog finishes first.
///
/// The wait itself is already a POLL (drained-or-deadline, below) — it returns
/// the instant every queue is empty and never sleeps a fixed span. The one
/// thing a fixed default cannot know is the box's LOAD: on a heavily loaded rig
/// a healthy standby's normal backlog can take longer than 30 s to drain, and
/// clearing the roster early truncates the flush. So the ceiling is TUNABLE —
/// `GUNNAR_ABDICATE_FLUSH_SECS` raises it for a loaded deployment without
/// changing the poll's shape (still drained-or-deadline, still short-circuits).
#[cfg(feature = "replication")]
const ABDICATE_FLUSH_SECS: u64 = 30;
/// The abdication flush ceiling in effect: `GUNNAR_ABDICATE_FLUSH_SECS` if it
/// parses to a positive value, else the [`ABDICATE_FLUSH_SECS`] default. A zero
/// or unparseable value is ignored (a 0 s ceiling would clear the roster before
/// a single job could drain — never what an operator means).
#[cfg(feature = "replication")]
fn abdicate_flush_secs() -> u64 {
// The raw string is HANDED IN to the pure resolver so the parse is unit
// tested without `set_var` — which is process-global and flakes the whole
// suite across threads (see `wg.rs`'s note on the same hazard).
resolve_flush_secs(std::env::var("GUNNAR_ABDICATE_FLUSH_SECS").ok().as_deref())
}
/// Pure ceiling resolver: a positive parsed override wins; anything else
/// (absent, blank, non-numeric, zero) falls back to [`ABDICATE_FLUSH_SECS`].
#[cfg(feature = "replication")]
fn resolve_flush_secs(raw: Option<&str>) -> u64 {
raw.and_then(|v| v.trim().parse::<u64>().ok())
.filter(|&s| s > 0)
.unwrap_or(ABDICATE_FLUSH_SECS)
}
/// Company SSO's storage on the identity table (`gunnar_grpc::sso::plane::SsoPlane`).
mod sso_plane;
/// **The refusal a read of the transaction log answers** — one spelling for `Mirror.LogFetch`
/// gathered and streamed: a released log is the named, retryable [`gunnar_grpc::PRIMARY_GROWING`],
/// a range retention dropped is `FailedPrecondition` (a snapshot is needed), a sequence above the
/// head is `InvalidArgument`.
fn log_read_refusal(e: gunnar_core::txlog::Error) -> ControlError {
match e {
gunnar_core::txlog::Error::Released(_) => ControlError::Unavailable(format!(
"{} this box's data set is being grown (unmounted): its transaction log is back after the remount — ask again",
gunnar_grpc::PRIMARY_GROWING
)),
gunnar_core::txlog::Error::NotRetained { seq, first } => ControlError::FailedPrecondition(format!(
"seq {seq} is no longer retained; the oldest is {first} — a snapshot is needed, not a range"
)),
gunnar_core::txlog::Error::BeyondHead { seq, head } => ControlError::InvalidArgument(format!("seq {seq} is beyond the head {head}")),
other => ControlError::Internal(format!("transaction log: {other}")),
}
}
/// **A streamed range onto the wire as `LogChunk` pieces** (twin-streaming, 2026-10-06): `meta`
/// on an entry's first piece, `last` on the piece that completes its declared length — which
/// [`crate::twin::stream_range`] hands over only after the entry's payload crc checked — and an
/// entry with no blob as one empty last piece at its end.
///
/// Since restore-large-entry (2026-10-07) every piece says the byte of the blob it starts at
/// (`offset`), an entry may begin part-way ([`crate::twin::stream_range_at`], the resumed
/// restore's `from_offset`), and the last piece carries the entry's checked payload crc
/// (`entry_crc`), so a box that put the entry together across several streams applies it only
/// if it IS the entry.
#[allow(clippy::type_complexity)] // the seam's sink, spelled as the seam spells it
struct ChunkOut<'s> {
sink: &'s mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
open: Option<OutEntry>,
}
/// The entry [`ChunkOut`] is sending.
struct OutEntry {
seq: u64,
kind: u32,
/// Not yet sent: it goes on the first piece.
meta: Option<Vec<u8>>,
blob_len: u64,
/// The byte of the blob the next piece starts at.
at: u64,
/// The checked payload crc, once the source said it (before the last piece).
crc: Option<u32>,
/// The `last` piece went out.
ended: bool,
}
impl ChunkOut<'_> {
fn send(&mut self, blob: &[u8]) -> anyhow::Result<()> {
let Some(o) = self.open.as_mut() else {
anyhow::bail!("a piece outside an entry");
};
let offset = o.at;
o.at += blob.len() as u64;
anyhow::ensure!(
o.at <= o.blob_len,
"seq {}: more than its {} blob bytes",
o.seq,
o.blob_len
);
let last = o.at == o.blob_len;
o.ended |= last;
let chunk = gunnar_proto::LogChunk {
seq: o.seq,
kind: o.kind,
meta: o.meta.take().unwrap_or_default(),
blob_len: o.blob_len,
blob: blob.to_vec(),
last,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: Some(offset),
entry_crc: if last { o.crc } else { None },
snapshot_item: None,
snapshot_id: String::new(),
};
(self.sink)(chunk).map_err(anyhow::Error::from)
}
}
impl crate::twin::FrameSink for ChunkOut<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
self.begin_at(seq, kind, meta, blob_len, 0)
}
fn begin_at(
&mut self,
seq: u64,
kind: u16,
meta: &[u8],
blob_len: u64,
offset: u64,
) -> anyhow::Result<()> {
anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
anyhow::ensure!(
offset <= blob_len,
"seq {seq}: from byte {offset} of a {blob_len}-byte blob"
);
self.open = Some(OutEntry {
seq,
kind: u32::from(kind),
meta: Some(meta.to_vec()),
blob_len,
at: offset,
crc: None,
ended: false,
});
Ok(())
}
fn checked(&mut self, payload_crc: u32) -> anyhow::Result<()> {
if let Some(o) = self.open.as_mut() {
o.crc = Some(payload_crc);
}
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
self.send(bytes)
}
fn end(&mut self) -> anyhow::Result<()> {
// An empty blob, or one resumed AT its end: one empty last piece carries the end (and
// the crc).
if matches!(&self.open, Some(o) if !o.ended) {
self.send(&[])?;
}
let Some(o) = self.open.take() else {
anyhow::bail!("an entry ended that never began");
};
anyhow::ensure!(
o.at == o.blob_len && o.ended,
"seq {}: {} of {} blob bytes went out",
o.seq,
o.at,
o.blob_len
);
Ok(())
}
}
impl CoreControlPlane {
/// **The log `Mirror.LogFetch` reads**, or the named refusal while it cannot be read: a set
/// held for its own growth (the guard over an unmounted set; MEASURED release-3 mock
/// 2026-09-29: `txlog: No such file or directory` mid twin-grow), or a PRIMARY whose log is
/// released for its own growth's remount (M1, release-3 test2 mock 2026-09-29) — both named
/// and retryable, never `Internal: txlog: No such file or directory`.
fn readable_log(&self) -> ControlResult<&gunnar_core::txlog::Log> {
if crate::twin::maintenance::is_held(self.stores.root()) {
return Err(ControlError::Unavailable(
"this box's data set is held for its own growth (unmounted): its transaction log is not readable until the growth mounts it back".into(),
));
}
if !self.stores.is_twin() {
if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
return Err(ControlError::FailedPrecondition(format!(
"{why}. Nothing is read from it"
)));
}
}
let log: &gunnar_core::txlog::Log = self.stores.txlog();
if log.is_released() {
return Err(log_read_refusal(gunnar_core::txlog::Error::Released(
log.dir().to_path_buf(),
)));
}
// ★ twin-resync-space (2026-10-06): a twin that gave its stale copy up for a resync, or
// is mid-install, holds no state at any one sequence — its log is not half a recharge.
if self.stores.is_twin() {
if let Some(why) = crate::twin::snapshot::not_whole(self.stores.root()) {
return Err(ControlError::FailedPrecondition(format!(
"{why}. Nothing is read from it"
)));
}
}
Ok(log)
}
}
/// The control plane, backed by the real catalog and identity table.
pub struct CoreControlPlane {
stores: Arc<ServerStores>,
/// **The runtime data-set growth** (`crate::grow`), when `serve` wired
/// one. `None` on a bare plane (tests, `--dry-run`): `GrowStatus` then
/// answers the seam's honest default and `GrowStart` is refused by name.
grow: Option<Arc<crate::grow::GrowMachine>>,
/// Started `--role twin` ([`CoreControlPlane::with_twin_role`]).
twin_role: bool,
/// **The git listener this server actually bound**, when `serve` handed it
/// in ([`CoreControlPlane::with_git_listener`]) — the fact
/// `Bootstrap.Status.serving_git` reports. `None` on a bare plane, and the
/// status door then falls back to the boot window alone, exactly as it
/// always answered. See [`crate::readiness::GitListener`].
git_listener: Option<crate::readiness::GitListener>,
/// **The ONE admission gate** the three git fronts share, when `serve`
/// handed it in — read by `Mirror.FlushReplication` for the in-flight
/// count the watermark needs (T11). `None` on a bare plane.
admission: Option<Arc<gunnar_ssh::Admission>>,
/// **What replication is doing**, for `Admin.Stats`.
///
/// A [`ReplicationView`](crate::replication_view::ReplicationView) and not
/// a replicator: this plane is built before the replicator is started
/// (replication is started last on purpose, so nothing above it can be
/// slowed by it), and the view is the slot `serve` drops it into
/// afterwards. It also keeps `feature = "replication"` and every replication type
/// out of this file, which is what lets the two lanes working on that
/// module and on this one stay out of each other's way.
///
/// Unfilled it reads `configured: false` — what a box with no standby *is*,
/// and a fact rather than an absence.
replication: crate::replication_view::ReplicationView,
/// **What replication is CONFIGURED to do**, for `Mirror.SetReplication` /
/// `RemoveReplication` / `ReplicationStatus`.
///
/// The write half beside the read half above, and the two are not twins: the
/// view answers *what has happened*, the slot decides *what runs*. It is
/// `Option` and behind the feature's own name only through
/// [`with_replication`](Self::with_replication), so a build without
/// `replication` — and a `CoreControlPlane` built by a test that has no
/// server around it — refuses the three verbs by name instead of failing to
/// compile.
///
/// **This is the whole reason the appliance can be told about a standby.**
/// the reference appliance is an OS-less image running as PID 1: no shell, no
/// editor, no `scp`. Before this, `replication.json` could only be written
/// by something with a filesystem to write it from, which on that box is
/// nothing at all.
#[cfg(feature = "replication")]
replication_slot: Option<Arc<crate::mirror::ReplicationSlot>>,
/// **The live webhook slot**, so an operator can name an outbound endpoint
/// over the control plane — the appliance has no shell, exactly as with the
/// replication slot above.
#[cfg(feature = "webhooks")]
webhook_slot: Option<Arc<crate::webhook::WebhookSlot>>,
/// **The live entitlement slot** (`metered` only): what `Entitlement.Set`
/// writes and the front doors read. `None` on a metered build started
/// without `--monetize-pubkey`: every namespace Free, and `Set` refuses by
/// name rather than accepting a fact it could not verify.
#[cfg(feature = "metered")]
entitlement: Option<Arc<crate::entitlement::EntitlementSlot>>,
/// **The disk-headroom gate** `Repos.Create` asks — the SAME object the
/// front doors' chain holds, so "the disk is full" is one line on both
/// doors (one implementation). `None` on a bare plane (a test not about the disk).
headroom: Option<Arc<gunnar_ssh::DiskHeadroomGate>>,
/// The largest object `Refs.ReadObject` will inflate. See
/// [`DEFAULT_MAX_OBJECT_BYTES`] for the arithmetic and
/// [`with_max_object_bytes`](Self::with_max_object_bytes) for changing it.
max_object_bytes: u64,
/// **The one gc at a time this box runs**, behind `Repos.Gc` and
/// `Repos.GcStatus`.
///
/// It belongs to the plane and not to a caller for the reason
/// [`crate::gc_queue`] opens with: a console can serialise itself, and
/// nothing else. Built here so every door onto this plane — the console, the
/// CLI over gRPC, anything else holding a credential — queues behind the
/// same worker. No thread is started until something is actually enqueued.
gc: crate::gc_queue::GcQueue,
/// **The one account purge at a time this box runs**, behind
/// `Accounts.Purge` and `Accounts.PurgeStatus`.
///
/// Its own queue and not the gc one — see [`crate::purge_queue`]: a gc
/// waits holding nothing, a purge waits holding a total lockout on a
/// customer's account, and putting them in one line would keep a user shut
/// out through somebody else's repository maintenance.
purge: crate::purge_queue::PurgeQueue,
/// **The one mirror sync at a time this box runs**, behind `Mirror.Sync`
/// and `Mirror.Status`.
///
/// Its own queue and not the gc one — see [`crate::mirror_queue`]: a gc
/// waits holding nothing, a mirror waits holding a repository that is not
/// yet usable, and making a first sync queue behind somebody else's hour of
/// compaction is the mistake `job_queue`'s module doc names in advance.
mirror: crate::mirror_queue::MirrorQueue,
/// **The directory import, one per account at a time** (`Sso.ImportUsers`
/// and `Sso.ImportStatus`, A34-17): its own lane, see
/// [`crate::sso_import_queue`].
sso_import: crate::sso_import_queue::SsoImportQueue,
/// **Where the event bus arrives, if this plane is ever given one.**
///
/// The mirror runner is built in [`CoreControlPlane::new`] and the bus is
/// handed over later by [`CoreControlPlane::with_bus`] (`serve.rs` builds
/// the bus first and the plane after, but `new` is also what every test
/// calls). A slot the closure captures is what lets the runner publish
/// `MirrorSynced` **when the sync finishes** rather than when it was asked
/// for — an event announcing a fetch that has not happened is worse than no
/// event.
///
/// `OnceLock`, so it is written once and read without a lock on every
/// completed sync. A plane that never gets one simply publishes nothing,
/// which is what every test plane wants.
bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
/// **The keys `Bootstrap.Restore` is taken from, as the medium named them**
/// (`--restore-pubkey`, [`CoreControlPlane::with_restore_keys`]). The installer's fed
/// keys are added at each ask ([`gunnar_grpc::ControlPlane::restore_keys`]). Empty on a
/// bare plane, which with no feed is a box that refuses every restore by name.
restore_keys: Vec<String>,
}
impl CoreControlPlane {
/// Serve out of `stores`, with the default object ceiling.
pub fn new(stores: Arc<ServerStores>) -> Self {
let for_gc = Arc::clone(&stores);
let bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>> = Arc::default();
CoreControlPlane {
stores,
grow: None,
twin_role: false,
git_listener: None,
admission: None,
replication: crate::replication_view::ReplicationView::new(),
#[cfg(feature = "replication")]
replication_slot: None,
#[cfg(feature = "webhooks")]
webhook_slot: None,
#[cfg(feature = "metered")]
entitlement: None,
headroom: None,
max_object_bytes: DEFAULT_MAX_OBJECT_BYTES,
gc: Self::gc_queue(Arc::clone(&for_gc)),
purge: Self::purge_queue(Arc::clone(&for_gc)),
mirror: Self::mirror_queue(Arc::clone(&for_gc), Arc::clone(&bus)),
sso_import: crate::sso_import_queue::SsoImportQueue::new(
crate::sso_import_queue::SsoImport,
),
bus,
restore_keys: Vec::new(),
}
}
/// **Keys this box takes a `Bootstrap.Restore` from** (h-v1-6a, 2026-09-28): OpenSSH
/// public key lines, from `--restore-pubkey`. For a box that no installer feeds — the
/// kit, a drill — this is the only way a restore can be taken at all.
pub fn with_restore_keys(mut self, keys: Vec<String>) -> Self {
self.restore_keys = keys;
self
}
/// **Publish this plane's event-stream events onto `bus`.**
///
/// Today that is one event and one producer: `MirrorSynced`, from the
/// mirror queue's worker, when a sync actually completes. It is a slot
/// rather than a constructor argument because `CoreControlPlane::new` is
/// what every test calls and a bus is not something a test of the disk
/// should have to build.
pub fn with_bus(self, bus: Arc<gunnar_grpc::EventBus>) -> Self {
// A second `with_bus` is the caller's mistake and is ignored rather
// than panicking a running server; the first bus wins, which is the
// one the listener is serving from.
let _ = self.bus.set(bus);
self
}
/// **This box was started `--role twin`** (`serve.rs`). `Bootstrap.Status`
/// then answers `role twin` before the join, when `twin.json` does not
/// exist yet and [`ServerStores::is_twin`] still says no — the moment an
/// installer door and a re-image verify ask.
pub fn with_twin_role(mut self, twin_role: bool) -> Self {
self.twin_role = twin_role;
self
}
/// **The address the git listener took**, handed in by `serve::start` from
/// `local_addr()` beside the address that was asked for.
///
/// This is what makes `Bootstrap.Status.serving_git` a reading of a SOCKET
/// rather than of a boot latch. Per-plane and not a static: two servers in
/// one test process are two listeners, and a process-wide slot would let
/// the second answer for the first.
pub fn with_git_listener(mut self, listener: crate::readiness::GitListener) -> Self {
self.git_listener = Some(listener);
self
}
/// **Release every purge lease this box is still holding.**
///
/// Called once at start-up, and it is the fast half of the crash-recovery
/// story. A purge that was interrupted by a SIGKILL or a power cut left its
/// lease in the identity table; nothing is running any more, so every one
/// of them is stale by construction — this process has just started and has
/// enqueued nothing.
///
/// Without it those accounts stay shut until their leases expire, which is
/// up to [`gunnar_core::PURGE_LEASE_MS`] of a paying customer being told
/// their account is locked by a purge that is not happening. The lease is
/// still there as the backstop for the case this cannot cover — a box that
/// never comes back up.
///
/// **Administrative gates are left alone.** They carry no deadline and were
/// set by a person; a restart must not un-suspend an account.
///
/// Returns how many were released. Failures are logged and not fatal: a
/// server that refused to start because it could not tidy a lease would
/// turn a recoverable state into an outage.
pub fn release_stale_purge_leases(stores: &ServerStores) -> u64 {
let gates = match stores.auth().gates() {
Ok(gates) => gates,
Err(err) => {
tracing::error!(
target: "gunnar::audit",
error = %err,
"the account gates could not be read at start-up; any purge lease left by a \
crash will stay in force until it expires"
);
return 0;
}
};
let mut released = 0;
for gate in gates {
if gate.reason != gunnar_core::GateReason::Purge {
continue;
}
match stores.auth().lift_gate_held_by(&gate.account, &gate.holder) {
Ok(Some(_)) => {
released += 1;
tracing::warn!(
target: "gunnar::audit",
event = "account.stale_lease_released",
account = %gate.account,
job = %gate.holder,
"an account was still locked by a purge that did not finish; the lockout \
has been lifted and the account may still hold bytes the purge did not \
reclaim — `repo list-orphans` shows them"
);
}
Ok(None) => {}
Err(err) => tracing::error!(
target: "gunnar::audit",
account = %gate.account,
error = %err,
"a stale purge lease could not be lifted at start-up"
),
}
}
released
}
/// The purge queue's two closures: the walk, and the unlock.
///
/// The walk is [`crate::account_purge::purge_account`] — the same function
/// the CLI verb calls, so there is one definition of what emptying an
/// account means (one implementation). The unlock is
/// `AuthTable::lift_gate_held_by`, **held_by** and not the unconditional
/// lift, so a purge finishing cannot clear a suspension an operator put on
/// while it ran.
///
/// The lease is re-stamped on every store reclaimed. A purge of a large
/// account outlives [`gunnar_core::PURGE_LEASE_MS`] easily, and a lease
/// that ran out mid-walk would unlock the account the walk is still
/// emptying — a reader could then start re-exploding a read cache into a
/// directory the purge has not reached yet. A renewal that FAILS is logged
/// and not fatal: it means the row is somebody else's now, the account is
/// still shut, and stopping the purge over it would leave more behind.
fn purge_queue(stores: Arc<ServerStores>) -> crate::purge_queue::PurgeQueue {
let for_release = Arc::clone(&stores);
crate::purge_queue::PurgeQueue::new(
Arc::new(
move |order: &crate::purge_queue::PurgeOrder,
progress: &crate::purge_queue::Progress| {
let request = crate::account_purge::PurgeRequest {
account: order.account.clone(),
principal_id: order.principal_id.clone(),
reason: order.reason.clone(),
remove_principal: order.remove_principal,
expect_empty: order.expect_empty,
};
let auth = stores.auth();
let account = order.account.clone();
let job_id = order.job_id.clone();
let mut step = |step: crate::account_purge::Step| match step {
crate::account_purge::Step::Found(stores_found) => {
progress.found(stores_found)
}
crate::account_purge::Step::Reclaimed { bytes, .. } => {
progress.reclaimed(bytes);
// **The heartbeat.** Pushed out after every store,
// so a purge that takes an hour never runs out of
// lease — and a purge that dies still hands the
// account back within PURGE_LEASE_MS.
let until = gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS;
if let Err(err) = auth.renew_gate(&account, &job_id, until) {
tracing::warn!(
target: "gunnar::audit",
account = %account,
job = %job_id,
error = %err,
"a purge could not renew its lockout lease; the account is \
still shut and the walk continues"
);
}
}
crate::account_purge::Step::Failed(note) => progress.failed(note),
crate::account_purge::Step::PrincipalRemoved => {
progress.principal_removed()
}
};
let report = crate::account_purge::purge_account(&stores, request, &mut step)
.map_err(|err| format!("{err:#}"))?;
// A walk that ran and left stores behind is a FAILURE, and
// the queue is told so here rather than left to infer it
// from a report it cannot see. The names are already on the
// job through `Step::Failed`; this is the sentence.
if report.failures.is_empty() {
Ok(())
} else {
Err(format!(
"{} of {} stores could not be reclaimed and are still on the disk; \
`repo list-orphans` shows them",
report.failures.len(),
report.stores_found
))
}
},
),
Arc::new(move |account: &str, job_id: &str| {
for_release
.auth()
.lift_gate_held_by(account, job_id)
.map(|lifted| lifted.is_some())
.map_err(|err| format!("{err}"))
}),
)
}
/// The queue's runner: resolve the catalog row, then **call the same
/// `admin::gc_one` the CLI verb calls** (one implementation).
///
/// Errors come back as strings because they are going onto a job a console
/// will read, not up a call stack. The resolution happens *inside* the
/// runner rather than at enqueue time on purpose: a repository can be
/// deleted while its job waits, and discovering that when the work starts
/// is a FAILED job saying so, which is the truth. Resolving early would
/// have queued a row that no longer describes anything.
fn gc_queue(stores: Arc<ServerStores>) -> crate::gc_queue::GcQueue {
crate::gc_queue::GcQueue::new(Arc::new(move |path: &str| {
let repo_ref = gunnar_core::resolve_repo_path(path)
.map_err(|err| format!("`{path}` is not a valid <namespace>/<name>: {err:#}"))?;
let repo = stores
.catalog()
.get(&repo_ref)
.map_err(|err| format!("reading the catalog row for {path}: {err:#}"))?
.ok_or_else(|| format!("{path} is no longer a live repository"))?;
let report = crate::admin::gc_one(&stores, &repo)
.map_err(|err| format!("maintaining {path}: {err:#}"))?;
Ok(crate::admin::reclaimed_bytes(&report))
}))
}
/// **The mirror queue's runner: read the sealed credential, then run the
/// one fetch (one implementation).**
///
/// The credential is read **here, on the worker**, and not carried on the
/// order. Two reasons, and both are about where a secret is allowed to be:
/// an order is `Debug` and lives on a job a status RPC renders, so a key on
/// it would be one `{:?}` away from a log line; and a credential rotated
/// while a sync waited should be the one the sync uses, not the one that
/// was current when somebody clicked.
///
/// Errors come back as strings because they are going onto a job a console
/// will read, not up a call stack — the same reason `gc_queue`'s runner
/// returns one. None of them can contain the credential: the only paths out
/// of `mirror_fetch` that mention the remote scrub its userinfo first.
fn mirror_queue(
stores: Arc<ServerStores>,
bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
) -> crate::mirror_queue::MirrorQueue {
crate::mirror_queue::MirrorQueue::new(Arc::new(
move |order: &crate::mirror_queue::MirrorOrder,
report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>| {
let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
format!(
"`{}` is not a valid <namespace>/<name>: {err:#}",
order.path
)
})?;
// **In flight, durably**, before anything is dialled: a restart
// during the sync leaves this mark, and the next start settles
// the sync as interrupted and resumes it
// (`resume_interrupted_mirrors`) instead of leaving a
// repository with nothing in it and a registration that says
// nothing went wrong.
let first = first_sync_of(
stores
.mirrors()
.mark_started(&repo_ref, gunnar_grpc::bus::now_unix_ms()),
&order.path,
)?;
let outcome = read_credential(&stores, &repo_ref).and_then(|credential| {
crate::mirror_fetch::sync_once(&stores, order, &credential, report)
});
audit_mirror_outcome(
&stores,
&repo_ref,
order,
first,
outcome.as_ref().err().map(String::as_str),
report,
);
// **The registration records every attempt, good or bad.**
// `record_sync` moves `last_sync_at_unix_ms` only on success —
// that field means "when this mirror was last correct", and
// moving it on a failure would make a mirror that has been
// broken for a week look fresh.
let error = outcome.as_ref().err().map(String::as_str);
if let Err(err) =
stores
.mirrors()
.record_sync(&repo_ref, gunnar_grpc::bus::now_unix_ms(), error)
{
tracing::warn!(
target: "gunnar::git",
repo = %order.path,
%err,
"a mirror sync finished and its registration could not be updated; the \
fetch itself is unaffected"
);
}
// The event stream, **after** the work and only on success. An event that
// announced a fetch which had not happened would be worse than
// no event, which is why `Mirror.Sync` no longer publishes one.
if outcome.is_ok() {
if let Some(bus) = bus.get() {
let job = report.snapshot();
let updates = job
.map(|job| job.report.updates)
.unwrap_or_default()
.into_iter()
.map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
ref_name,
old_oid,
new_oid,
})
.collect::<Vec<_>>();
bus.publish(gunnar_proto::event::Payload::MirrorSynced(
gunnar_proto::MirrorSynced {
repo: Some(RepoPath::new(repo_ref.namespace(), repo_ref.name())),
remote_url: order.upstream.clone(),
refs_changed: updates.len() as u32,
updates,
},
));
}
}
outcome
},
))
}
/// **Settle every sync a stopped process left in flight, and resume it**
/// (import-big, 2026-10-26).
///
/// The job queue is memory: a restart in the middle of an import (an
/// out-of-memory kill, an update, a power cut) forgot the job, and what was
/// left was a repository with nothing in it and a registration with no
/// error — a console that showed no failure for an import that never
/// landed. The registration now carries an in-flight mark
/// (`MirrorRecord::sync_started_at_unix_ms`); here, once at start, each
/// marked sync is settled as interrupted (its `last_error` says so, and the
/// account's audit log has a line) and enqueued again, up to
/// [`gunnar_core::MAX_RESUMES`] times in a row. Past that it stays FAILED
/// with the reason, for a person to press Sync. Returns how many were
/// resumed.
pub fn resume_interrupted_mirrors(&self) -> usize {
let settled = match self
.stores
.mirrors()
.settle_interrupted(gunnar_grpc::bus::now_unix_ms())
{
Ok(settled) => settled,
Err(err) => {
tracing::warn!(target: "gunnar::git", %err, "the mirror registrations could not be read to resume interrupted syncs");
return 0;
}
};
let mut resumed = 0;
for record in settled {
let Ok(repo_ref) = gunnar_core::resolve_repo_path(&record.repo) else {
continue;
};
let again = record.interrupted <= gunnar_core::MAX_RESUMES;
audit_line(
&self.stores,
&repo_ref,
if again {
"mirror.sync.interrupted"
} else {
"mirror.sync.failed"
},
&record.last_error,
true,
);
tracing::warn!(
target: "gunnar::git",
event = "mirror.sync.interrupted",
repo = %record.repo,
interrupted = record.interrupted,
resumed = again,
"a mirror sync was in flight when the server stopped"
);
if again {
self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
path: repo_ref.key(),
upstream: record.remote_url.clone(),
prune: record.prune,
ca: mirror_ca_of(&record),
});
resumed += 1;
}
}
resumed
}
/// **Take the lockout a purge runs under, without downgrading one that is
/// already there.**
///
/// A lease is written only when the account is *not* already gated. If it
/// is — an operator suspending a customer, monetize locking a non-payer —
/// that gate stands, and the purge runs under it. Two consequences, both
/// deliberate:
///
/// * the account does not quietly acquire a **deadline** it did not have.
/// Overwriting an administrative gate with a lease would mean a
/// suspension expiring ten minutes after a purge finished with it, which
/// is a suspended account silently coming back.
/// * the job will report `gate_released: false` and the account stays shut,
/// because `lift_gate_held_by` will find a holder that is not this job.
/// That is the correct outcome: the purge did not take that lockout and
/// has no standing to lift it.
fn gate_for_purge(
&self,
account: &str,
job_id: &str,
options: &gunnar_grpc::seam::PurgeAccountOptions,
) -> ControlResult<()> {
if self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.is_some()
{
return Ok(());
}
let lease = gunnar_core::AccountGate::purge_lease(
account,
job_id,
gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS,
)
.by(&options.principal_id)
.noted(format!(
"this account is being permanently deleted: {}",
options.reason
));
self.stores.auth().gate(lease).map_err(gate_error)?;
Ok(())
}
/// **Hand the disk-headroom gate in.** From here on `Repos.Create` refuses
/// under the floor with `RESOURCE_EXHAUSTED` and the gate's own line —
/// the one a refused `git push` prints, because it is the same function.
#[must_use]
pub fn with_headroom(mut self, gate: Arc<gunnar_ssh::DiskHeadroomGate>) -> Self {
self.headroom = Some(gate);
self
}
/// **Hand this plane the runtime growth** (`crate::grow`). Called by
/// `serve` once the admission gate and the stores exist.
pub fn with_grow(mut self, grow: Arc<crate::grow::GrowMachine>) -> Self {
self.grow = Some(grow);
self
}
/// **Hand this plane the admission gate** the git fronts share, for the
/// in-flight count `Mirror.FlushReplication` reports.
pub fn with_admission(mut self, admission: Arc<gunnar_ssh::Admission>) -> Self {
self.admission = Some(admission);
self
}
/// The growth, or the seam's refusal by name.
fn grow_machine(&self) -> ControlResult<&Arc<crate::grow::GrowMachine>> {
self.grow.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no runtime growth wired in, so it cannot grow the data set",
))
}
/// Verify a signed membership event before it changes anything.
///
/// The same shape as `forge_append`'s check and for the same reason (one
/// implementation): a membership log that took an event it could not verify would be a
/// log of claims, and the point of the design is that it is a log of
/// evidence. The key must be in the ACL table (or no later reader could
/// check it), it must be the claimed actor's own, and the event must
/// describe the change the caller named — re-checked here at the seam
/// because this door can also be reached without a connection.
fn verify_membership_event(
&self,
signed: gunnar_proto::SignedMembershipEvent,
group_id: &str,
member_id: &str,
action: gunnar_core::MembershipAction,
) -> ControlResult<gunnar_core::SignedMembershipEvent> {
let signed = from_proto_membership(signed)?;
if signed.event.group != group_id || signed.event.member != member_id {
return Err(ControlError::InvalidArgument(format!(
"the signed event is about {:?}/{:?} but the change is {:?}/{:?}",
signed.event.group, signed.event.member, group_id, member_id
)));
}
if signed.event.action != action {
return Err(ControlError::InvalidArgument(
"the signed event's action does not match the change (add vs remove)".to_owned(),
));
}
let key = self
.stores
.auth()
.signing_keys()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.into_iter()
.find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
.ok_or_else(|| {
ControlError::InvalidArgument(format!(
"{} is not registered on this server, so nobody reading this event later \
could check its signature",
signed.key_fingerprint
))
})?;
if key.principal_id != signed.event.actor {
return Err(ControlError::InvalidArgument(format!(
"this membership change is attributed to {:?} and is signed by {}, which is \
{:?}'s key; a membership change is the actor's own word, so it must carry the \
actor's own signature",
signed.event.actor, signed.key_fingerprint, key.principal_id
)));
}
let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
.map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
gunnar_core::membership::verify_event(&signed, &public)
.map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
Ok(signed)
}
/// Append a verified membership event to the signed membership log, and
/// return the index it landed at — the index an inclusion proof is taken
/// at. Append-only and never rewritten, so a removal followed by a re-add
/// leaves two standing events, which is what makes the history evidence.
fn append_membership_event(
&self,
signed: &gunnar_core::SignedMembershipEvent,
) -> ControlResult<u64> {
let log = self.stores.membership_log();
let mut line = serde_json::to_string(signed)
.map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
line.push('\n');
let _framed =
gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
.map_err(|err| {
ControlError::Unavailable(format!(
"framing the line in the transaction log: {err}"
))
})?;
use std::io::Write as _;
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(log.records_path())
.map_err(|err| {
ControlError::Unavailable(format!("opening the membership log: {err}"))
})?;
file.write_all(line.as_bytes()).map_err(|err| {
ControlError::Unavailable(format!("appending to the membership log: {err}"))
})?;
let count = log
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.len() as u64;
Ok(count.saturating_sub(1))
}
/// Every membership event on the server, decoded, in append order.
fn membership_log_events(&self) -> ControlResult<Vec<gunnar_core::SignedMembershipEvent>> {
self.stores
.membership_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.iter()
.map(|line| {
serde_json::from_str(line).map_err(|err| {
ControlError::Internal(format!("decoding a membership event: {err}"))
})
})
.collect()
}
/// **Hand this plane the live replication slot**, so an operator can name a
/// standby over the control plane.
///
/// Called by `Server::start` before the plane is wrapped in an `Arc`, and
/// before anything is running in the slot — which is the point: a target has
/// to be settable on a box that has never had one and holds no repositories
/// yet, and that is every box on its first boot.
#[cfg(feature = "replication")]
pub fn with_replication(mut self, slot: Arc<crate::mirror::ReplicationSlot>) -> Self {
self.replication_slot = Some(slot);
self
}
/// The slot, or a refusal that names WHY there is none.
///
/// `Unimplemented` and not `Internal`: a plane built without a slot is a
/// plane with no server around it (a unit test), and telling a caller "this
/// server cannot do that" is true, actionable and not an incident.
#[cfg(feature = "replication")]
fn replication_slot(&self) -> ControlResult<&Arc<crate::mirror::ReplicationSlot>> {
self.replication_slot.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no replication slot, so it cannot configure a standby",
))
}
/// **Hand this plane the live webhook slot**, so an operator can name an
/// outbound endpoint over the control plane. Called by `Server::start`
/// before the plane is wrapped in an `Arc`.
#[cfg(feature = "webhooks")]
pub fn with_webhooks(mut self, slot: Arc<crate::webhook::WebhookSlot>) -> Self {
self.webhook_slot = Some(slot);
self
}
/// Hand the entitlement slot in (`metered` only).
#[cfg(feature = "metered")]
pub fn with_entitlement(mut self, slot: Arc<crate::entitlement::EntitlementSlot>) -> Self {
self.entitlement = Some(slot);
self
}
/// The entitlement slot, or a refusal that names WHY there is none.
#[cfg(feature = "metered")]
fn entitlement(&self) -> ControlResult<&Arc<crate::entitlement::EntitlementSlot>> {
self.entitlement.as_ref().ok_or_else(|| {
ControlError::FailedPrecondition(
"this metered server was started without `--monetize-pubkey`, so it holds no \
key to verify an entitlement fact against and refuses every one; every \
namespace is Free"
.into(),
)
})
}
/// The webhook slot, or a refusal that names WHY there is none.
#[cfg(feature = "webhooks")]
fn webhook_slot(&self) -> ControlResult<&Arc<crate::webhook::WebhookSlot>> {
self.webhook_slot.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no webhook slot, so it cannot configure an outbound endpoint",
))
}
/// Configuration **and** running state, in one answer.
///
/// Both halves, always, because they can disagree and the disagreement is
/// the interesting part: `targets` non-empty with `standbys` empty is a
/// server that has the configuration and is not running it.
#[cfg(feature = "replication")]
///
/// **`include_refs` is the operator's, and it is false for everything but a
/// `replication list --refs`.** The mutating verbs print a snapshot back to
/// confirm the roster, and the ref-coverage walk they used to trigger opens
/// every repository's reference database for a number nobody asked
/// for — so they pass `false`, and only the explicit status verb forwards
/// the flag the operator set.
fn replication_snapshot(
&self,
include_refs: bool,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let mut out = gunnar_proto::ReplicationStatusResponse {
targets: config
.targets
.iter()
.map(crate::replication_wire::target_to_proto)
.collect(),
role: config.role.to_proto() as i32,
..Default::default()
};
if let Some(replicator) = slot.replicator() {
crate::replication_wire::running_to_proto(&replicator, include_refs, &mut out);
}
Ok(out)
}
/// **The slot `serve` drops the replicator's counters into**, so
/// `Admin.Stats` can report them.
///
/// Without it the counters exist and nothing reads them, which is exactly
/// the state that let every configured standby receive NOTHING from
/// `e31ab8f4` to 2026-08-30 while five log lines out of five said so into a
/// file nobody was tailing.
pub fn replication_view(&self) -> crate::replication_view::ReplicationView {
self.replication.clone()
}
/// Every forge event, as it was written.
///
/// Verifying nothing: the fold is over what is in the log, and whether an
/// event is evidence is a separate question with its own answer, asked
/// per approval in [`CoreControlPlane::thread_state`].
fn forge_events(&self) -> ControlResult<Vec<gunnar_core::SignedEvent>> {
self.stores
.forge_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("reading the forge log: {err:#}")))?
.iter()
.map(|line| {
serde_json::from_str(line).map_err(|err| {
ControlError::Internal(format!("a forge event will not decode: {err}"))
})
})
.collect()
}
/// **The subject of a forge event must exist and be open** — the one check
/// for an event its actor signed and one gunnar attested (GHUNT-19).
fn forge_subject_open(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<()> {
// **The subject must exist and be open**, and which subject that is
// depends on the kind. A review event names a pull request the record
// owns; an issue event names an issue THIS log owns. Both refuse an
// event on something merged, closed or never opened — evidence about
// nothing — but they ask different authorities, so they branch here.
let (namespace, name) = signed.event.repo.split_once('/').ok_or_else(|| {
ControlError::InvalidArgument(format!(
"`repo` must be `<namespace>/<repo>`, got {:?}",
signed.event.repo
))
})?;
let path = RepoPath::new(namespace, name);
match &signed.event.event {
// An opening creates the issue and so has no precondition: there is
// nothing yet to be open. Its number is the fold's to assign.
gunnar_core::Event::IssueOpened { .. } => {}
// A comment or close names an issue by number. The issue is folded
// from this same log — there is no record behind it — so the check
// is against the fold rather than `get_pull`.
gunnar_core::Event::IssueCommented { .. } | gunnar_core::Event::IssueClosed { .. } => {
let repo_events: Vec<gunnar_core::ForgeEvent> = self
.forge_events()?
.into_iter()
.filter(|e| e.event.repo == signed.event.repo)
.map(|e| e.event)
.collect();
match gunnar_core::forge::fold_issue(signed.event.pull, &repo_events) {
None => {
return Err(ControlError::FailedPrecondition(format!(
"no issue #{} in {}: nobody has opened one, so an event on it is \
evidence about nothing",
signed.event.pull, signed.event.repo
)));
}
Some(issue) if !issue.open => {
return Err(ControlError::FailedPrecondition(format!(
"issue #{} in {} is closed; an event on it cannot change what has \
already happened",
signed.event.pull, signed.event.repo
)));
}
Some(_) => {}
}
}
// A review event names a pull request, whose record is the authority
// for whether it exists and is open.
gunnar_core::Event::Commented { .. }
| gunnar_core::Event::Approved { .. }
| gunnar_core::Event::Rejected { .. } => {
let record = self.get_pull(&path, signed.event.pull)?;
if record.state != gunnar_proto::PullRequestState::Open as i32 {
return Err(ControlError::FailedPrecondition(format!(
"pull request #{} in {} is not open; a review event cannot change what \
has already happened",
signed.event.pull, signed.event.repo
)));
}
}
}
Ok(())
}
/// Append a checked forge event to the log, and say where it landed.
fn forge_write(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<u64> {
let log = self.stores.forge_log();
let mut line = serde_json::to_string(signed)
.map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
line.push('\n');
let _framed =
gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
.map_err(|err| {
ControlError::Unavailable(format!(
"framing the line in the transaction log: {err}"
))
})?;
use std::io::Write as _;
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(log.records_path())
.map_err(|err| ControlError::Unavailable(format!("opening the forge log: {err}")))?;
file.write_all(line.as_bytes()).map_err(|err| {
ControlError::Unavailable(format!("appending to the forge log: {err}"))
})?;
// The index it landed at, counted after the write so it names this
// event and not the one before it.
let count = self
.stores
.forge_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.len() as u64;
Ok(count.saturating_sub(1))
}
/// A folded review, with **every approval re-checked against the ACL key
/// table right now** — not trusted because it is in the file.
///
/// That re-check is the difference the whole design is about, and it is why
/// the wire carries a verdict per approval rather than a count: an approval
/// whose key was since removed has not failed and has not passed, and a
/// number cannot say so.
///
/// **The key must be the claimed actor's own.** A signature proves which
/// KEY spoke; only the key table's `principal_id` says whose voice that
/// is. Until 2026-08-31 the fingerprint was looked up across the whole
/// table, so an approval claiming `actor: alice` signed by any other
/// registered principal's key verified — and merged. The verdict for that
/// case names the key's real owner, because "your signature failed" and
/// "that is somebody else's key" send an operator down different roads.
fn review_state(
&self,
repo: &str,
state: &gunnar_core::forge::Review,
all: &[gunnar_core::SignedEvent],
) -> ControlResult<gunnar_proto::ReviewState> {
let keys: Vec<(String, String, gunnar_core::ssh_key::PublicKey)> = self
.stores
.auth()
.signing_keys()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.iter()
.filter_map(|row| {
gunnar_core::ssh_key::PublicKey::from_openssh(&row.openssh)
.ok()
.map(|key| {
(
row.fingerprint_sha256.clone(),
row.principal_id.clone(),
key,
)
})
})
.collect();
// Gunnar's own forge-attestation key, read without minting: an
// attested approval verifies against it and nothing else (GHUNT-19).
let attest_key = self
.stores
.forge_attest_public()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
let approvals = state
.approvals
.iter()
.map(|(actor, commit)| {
let signed = all.iter().rev().find(|e| {
e.event.repo == repo
&& e.event.pull == state.pull
&& e.event.actor == *actor
&& matches!(&e.event.event, gunnar_core::Event::Approved { commit: c } if c == commit)
});
let (verified, verdict) = match signed {
None => (false, "NO EVENT".to_owned()),
// **Gunnar's attestation for a person's stepped-up
// session**: signed by gunnar's own key, the actor named,
// the step-up inside the signature. Checked against that
// key alone — never against a principal's.
Some(signed) if signed.attested.is_some() => match &attest_key {
None => (false, "UNKNOWN KEY".to_owned()),
Some(key) => match gunnar_core::forge::verify_event(signed, key) {
Ok(()) => (
true,
signed.attested.as_ref().map(|a| a.words()).unwrap_or_default(),
),
Err(problem) => (false, format!("BAD ({problem})")),
},
},
Some(signed) => match keys
.iter()
.find(|(fingerprint, _, _)| *fingerprint == signed.key_fingerprint)
{
None => (false, "UNKNOWN KEY".to_owned()),
// The key is registered — to somebody else. The
// signature may be perfectly good; it is still not
// this actor's word, and saying whose it is lets the
// operator see a filed-in-another's-name attempt for
// what it is.
Some((_, owner, _)) if *owner != *actor => {
(false, format!("KEY OF {owner}, NOT {actor}"))
}
Some((_, _, key)) => match gunnar_core::forge::verify_event(signed, key) {
Ok(()) => (true, "verified".to_owned()),
Err(problem) => (false, format!("BAD ({problem})")),
},
},
};
gunnar_proto::Approval {
actor: actor.clone(),
commit: commit.clone(),
verified,
verdict,
}
})
.collect();
Ok(gunnar_proto::ReviewState {
repo: repo.to_owned(),
pull: state.pull,
approvals,
comments: state.comments as u64,
})
}
/// **Set the deployment's object ceiling** for `Refs.ReadObject`.
///
/// `0` disables the ceiling, which restores the unbounded behaviour: the
/// server will inflate an object of any size to answer a request. That is a
/// real choice for a single-tenant box with a trusted control port and a
/// repository full of large blobs, and it is spelled explicitly rather than
/// being what you get by forgetting to configure anything.
///
/// This exists as much for the tests as for the operator: a ceiling that can
/// only be a compile-time constant can only be exercised by writing an object
/// bigger than it, which for the real default means a 64 MiB fixture. Being
/// able to set it small is what lets the refusal be seen red cheaply — and it
/// proves the configuration is *wired*, which a test against the constant
/// could never show.
pub fn with_max_object_bytes(mut self, bytes: u64) -> Self {
self.max_object_bytes = bytes;
self
}
/// The pull-request table, opened against the same control store the
/// catalog and the identity table live in. A fresh handle per call: the
/// document cache rides the handle, and a pull-request operation is rare
/// enough that re-parsing beats holding one more long-lived Arc chain.
fn pulls(&self) -> ControlResult<gunnar_core::Pulls> {
gunnar_core::Pulls::new(Arc::clone(self.stores.control_store()))
.map_err(|err| ControlError::Internal(format!("opening the pulls table: {err:#}")))
}
fn repo_ref(path: &RepoPath) -> ControlResult<RepoRef> {
path.validate()
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
RepoRef::new(&path.namespace, &path.repo)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))
}
/// Resolve a tree-ish — a ref name, a hex oid, or `HEAD` — to the **tree**
/// it names.
///
/// The resolution half is the one `log` already does: a hex oid is taken as
/// one, anything else is validated as a reference name and looked up in the
/// snapshot, and an oid of the wrong hash kind is refused with a sentence
/// that says which kind the repository is. What is added here is the last
/// hop — a commit is followed to its tree, a tag is peeled first, and a tree
/// is already the answer — so `from`/`to` accept everything `git diff`
/// accepts in the same position.
fn tree_of(
&self,
git: &dyn GitServe,
hash: gix_hash::Kind,
path: &RepoPath,
spec: &str,
) -> ControlResult<gix_hash::ObjectId> {
let id = match gix_hash::ObjectId::from_hex(spec.as_bytes()) {
Ok(id) => {
if id.kind() != hash {
return Err(ControlError::InvalidArgument(format!(
"{spec} is a {} oid; this repository is {hash}",
id.kind()
)));
}
id
}
Err(_) => {
gunnar_store::refname::validate(spec)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let resolved = if spec == gunnar_store::HEAD {
snapshot.head(git).map_err(wire_error)?
} else {
snapshot.resolve(spec).map_err(wire_error)?
};
resolved.ok_or_else(|| ControlError::NotFound(format!("{spec} in {path}")))?
}
};
let id = peel(git, hash, id)?;
let (kind, bytes) = git
.read(id.as_bytes())
.map_err(store_error)?
.ok_or_else(|| ControlError::NotFound(format!("object {id} in {path}")))?;
match kind {
ObjType::Tree => Ok(id),
ObjType::Commit => object::commit(&bytes, hash)
.map(|commit| commit.tree)
.map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}"))),
// A blob is a legal object and not a legal side of a tree diff.
// Saying so beats walking it and reporting that nothing changed.
other => Err(ControlError::InvalidArgument(format!(
"{spec} names a {other:?}; a diff needs a commit, a tag or a tree"
))),
}
}
/// The repository's policy as `gunnar-policy` understands it.
///
/// A repository with no stored policy reads as
/// [`PolicyConfig::default`](gunnar_policy::PolicyConfig) — no gates —
/// which is the same thing the push path concludes and is stated in one
/// place rather than two. A stored document that does **not parse** is an
/// error here and not an empty policy: this path is a caller asking what
/// the rules are, and answering "there are none" about a document that
/// exists and is broken is the worst available answer.
fn policy_of(&self, repo: &gunnar_core::Repo) -> ControlResult<gunnar_policy::PolicyConfig> {
let stored = self.stores.policies().get(&repo.path()).map_err(internal)?;
match stored {
None => Ok(gunnar_policy::PolicyConfig::default()),
Some(document) => gunnar_policy::PolicyConfig::from_json(&document).map_err(|err| {
ControlError::Internal(format!(
"the stored policy for {} does not parse: {err}",
repo.key()
))
}),
}
}
/// Write `config` back, or delete the document when it configures nothing.
///
/// Deleting rather than storing an empty document is what keeps "nobody has
/// configured this repository" and "somebody configured nothing" the same
/// state. Two states that behave identically and read differently are a
/// question an operator cannot answer.
fn store_policy(
&self,
repo: &gunnar_core::Repo,
config: &gunnar_policy::PolicyConfig,
) -> ControlResult<()> {
let policies = self.stores.policies();
if *config == gunnar_policy::PolicyConfig::default() {
policies.remove(&repo.path()).map_err(internal)?;
return Ok(());
}
let json = config
.to_json()
.map_err(|err| ControlError::Internal(format!("rendering the policy: {err}")))?;
policies.set(&repo.path(), &json).map_err(internal)
}
fn open(&self, path: &RepoPath) -> ControlResult<(gunnar_core::Repo, Arc<OpenRepo>)> {
let repo_ref = Self::repo_ref(path)?;
let repo = self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
let store = self.stores.repo_store(&repo).map_err(|err| {
ControlError::Internal(format!("opening the repository's stores: {err:#}"))
})?;
Ok((repo, store))
}
/// **A mirror's registration, with its repository opened when it has
/// one.** A registration whose repository is still to come (an import's
/// first sync has not landed: `MirrorRecord::creates_repo`) has none to
/// open, and that is not an error; every other registration's repository
/// must open, as it always had to.
fn mirror_record(&self, path: &RepoPath) -> ControlResult<gunnar_core::MirrorRecord> {
let repo_ref = Self::repo_ref(path)?;
let record = self.stores.mirrors().get(&repo_ref).map_err(internal)?;
match record {
Some(record)
if record.creates_repo
&& self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_none() =>
{
Ok(record)
}
record => {
self.open(path)?;
record.ok_or_else(|| {
ControlError::NotFound(format!("no mirror registered for {path}"))
})
}
}
}
}
/// **The one place a [`crate::gc_queue::GcJob`] becomes a seam job.**
///
/// The queue keys on `<namespace>/<name>` because that is what its runner
/// needs; the seam speaks [`RepoPath`] because that is what the wire does.
/// `path` is threaded in rather than re-parsed out of the job's own string: it
/// is the request's own path, already validated, and re-parsing it here would
/// be a second resolver that could disagree with the first (one implementation).
fn gc_job_out(path: &RepoPath, job: crate::gc_queue::GcJob) -> gunnar_grpc::seam::GcJob {
gunnar_grpc::seam::GcJob {
job_id: job.job_id,
path: path.clone(),
state: match job.state {
crate::gc_queue::GcState::Queued => gunnar_grpc::seam::GcState::Queued,
crate::gc_queue::GcState::Running => gunnar_grpc::seam::GcState::Running,
crate::gc_queue::GcState::Done => gunnar_grpc::seam::GcState::Done,
crate::gc_queue::GcState::Failed => gunnar_grpc::seam::GcState::Failed,
},
queue_position: job.queue_position,
// Named through `report` rather than through the job's `Deref`: the
// job is being taken apart field by field here, and a deref borrow in
// the middle of that would be reading a value that is half moved.
bytes_reclaimed: job.report.bytes_reclaimed,
error: job.error,
enqueued_at_unix_ms: job.enqueued_at_unix_ms,
finished_at_unix_ms: job.finished_at_unix_ms,
}
}
/// `gunnar_core`'s gate onto the seam's.
fn gate_out(gate: gunnar_core::AccountGate) -> gunnar_grpc::seam::AccountGate {
gunnar_grpc::seam::AccountGate {
account: gate.account,
reason: match gate.reason {
gunnar_core::GateReason::Administrative => {
gunnar_grpc::seam::GateReason::Administrative
}
gunnar_core::GateReason::Purge => gunnar_grpc::seam::GateReason::Purge,
},
holder: gate.holder,
actor: gate.actor,
note: gate.note,
at_unix_ms: gate.at,
// **0 on the seam means "no deadline"**, which is what `None` means
// here. Flattened at this boundary rather than on the wire so the proto
// stays a plain `int64` and every reader of it applies the same rule.
expires_unix_ms: gate.expires.unwrap_or(0),
}
}
/// The queue's job onto the seam's.
fn purge_job_out(job: crate::purge_queue::PurgeJob) -> gunnar_grpc::seam::PurgeJob {
// The purge's own half, taken out whole before the queue's half is: the
// failure list is owned and has to be MOVED onto the seam, and a job's
// `Deref` only ever lends its report.
let report = job.report;
gunnar_grpc::seam::PurgeJob {
job_id: job.job_id,
// The queue keys on the account, so the account IS the job's key.
account: job.key,
state: match job.state {
crate::purge_queue::PurgeState::Queued => gunnar_grpc::seam::PurgeState::Queued,
crate::purge_queue::PurgeState::Running => gunnar_grpc::seam::PurgeState::Running,
crate::purge_queue::PurgeState::Done => gunnar_grpc::seam::PurgeState::Done,
crate::purge_queue::PurgeState::Failed => gunnar_grpc::seam::PurgeState::Failed,
},
queue_position: job.queue_position,
stores_found: report.stores_found,
stores_purged: report.stores_purged,
bytes_reclaimed: report.bytes_reclaimed,
principal_removed: report.principal_removed,
gate_released: report.gate_released,
error: job.error,
failures: report.failures,
enqueued_at_unix_ms: job.enqueued_at_unix_ms,
started_at_unix_ms: job.started_at_unix_ms,
finished_at_unix_ms: job.finished_at_unix_ms,
}
}
/// A gate write that was refused, as the error the caller wants.
///
/// A reserved or malformed account name is the caller's mistake and comes back
/// as `INVALID_ARGUMENT`; anything else is the table failing to be written and
/// is not the caller's fault. Folding the first into `Internal` would send an
/// operator looking at the disk over a typo.
fn gate_error(err: gunnar_core::Error) -> ControlError {
match err {
gunnar_core::Error::ReservedPrincipalId(_) | gunnar_core::Error::InvalidPath { .. } => {
ControlError::InvalidArgument(format!("{err}"))
}
other => internal(other),
}
}
/// **One stored flow row → one `DayActivity`.** The single writer of that
/// mapping, so the wire and the ledger cannot drift by a field.
///
/// `refused_by_reason` carries only the reasons that HAPPENED. A reason with no
/// refusals that day is absent rather than 0, because the map is a breakdown of
/// a number the caller already has (`refused`) and a zero entry in it adds a
/// row to a console's list that says nothing.
fn day_activity(day: i64, row: &gunnar_core::FlowDay) -> gunnar_proto::DayActivity {
gunnar_proto::DayActivity {
day_unix_ms: gunnar_core::midnight_of(day),
bytes_in: row.bytes_in,
bytes_out: row.bytes_out,
pushes: row.pushes,
fetches: row.fetches,
lfs_bytes_in: row.lfs_bytes_in,
lfs_bytes_out: row.lfs_bytes_out,
refused: row.refused(),
refused_by_reason: gunnar_core::Refusal::ALL
.into_iter()
.filter(|reason| row.refused_for(*reason) > 0)
.map(|reason| (reason.name().to_owned(), row.refused_for(reason)))
.collect(),
auth_failures: row.auth_failures,
last_push_unix_ms: row.last_push_unix_ms,
last_fetch_unix_ms: row.last_fetch_unix_ms,
distinct_principals: row.distinct_principals(),
distinct_principals_capped: row.principals_capped,
}
}
/// **A create's refusal, classified** — the repository or namespace named, a box being
/// restored (or off the stripe) as UNAVAILABLE (RUN A 2026-10-08: a create on a main being
/// restored answered INTERNAL, and the console showed "transaction log" and "twin" to a
/// customer), any other core error INTERNAL, anything else (the disk's own refusal) as is.
pub(crate) fn create_error(err: &anyhow::Error, path: &RepoPath) -> ControlError {
match err.downcast_ref::<gunnar_core::Error>() {
Some(gunnar_core::Error::RepoExists(_)) => {
ControlError::AlreadyExists(format!("repository {path}"))
}
Some(gunnar_core::Error::NoSuchNamespace(ns)) => {
ControlError::NotFound(format!("namespace {ns}"))
}
Some(e @ (gunnar_core::Error::LogFollows | gunnar_core::Error::OffStripe { .. })) => {
ControlError::Unavailable(e.to_string())
}
Some(_) => ControlError::Internal(format!("{err:#}")),
None => ControlError::FailedPrecondition(format!("{err:#}")),
}
}
/// **Is this the mirror's first sync, from its in-flight mark?** (`mark_started`, IMPORT-BIG).
/// The mark is what lets a restart resume the import and what makes its outcome audited.
pub(crate) fn first_sync_of<E: std::fmt::Display>(
mark: Result<Option<gunnar_core::MirrorRecord>, E>,
path: &str,
) -> Result<bool, String> {
match mark {
Ok(record) => {
Ok(record.is_some_and(|record| record.creates_repo || record.sync_count == 0))
}
Err(err) => {
// RA4-10: unwritten, a restart could not resume this import and its outcome would go
// unaudited, so the sync does not start.
tracing::error!(target: "gunnar::git", "ALERT mirror.mark-unwritten {path}: the in-flight mark was not written ({err}); the sync is not started");
Err(format!("{path}: the mirror's in-flight mark was not written ({err}); nothing was fetched, ask again"))
}
}
}
#[cfg(test)]
mod mirror_mark_tests {
use super::*;
/// ★ RA4-10 RED: **an in-flight mark that is not written fails the sync by name** — a
/// dropped write meant a restart could not resume the import and its outcome went
/// unaudited.
#[test]
fn a_mark_that_is_not_written_fails_the_sync_by_name() {
let why = first_sync_of::<String>(Err("the mirrors store is full".into()), "acme/linux")
.expect_err("refused");
assert!(
why.contains("acme/linux")
&& why.contains("the mirrors store is full")
&& why.contains("in-flight mark"),
"{why}"
);
assert_eq!(first_sync_of::<String>(Ok(None), "acme/linux"), Ok(false));
}
}
#[cfg(test)]
mod create_error_tests {
use super::*;
/// ★ **A create on a box being restored is UNAVAILABLE, never INTERNAL** (RUN A R8,
/// 2026-10-08): the console turns UNAVAILABLE + this sentence into its own word.
#[test]
fn a_create_on_a_restoring_box_is_unavailable() {
let path = RepoPath::new("acme", "site");
let restoring = anyhow::Error::new(gunnar_core::Error::LogFollows);
assert!(
matches!(create_error(&restoring, &path), ControlError::Unavailable(m) if m.contains("being restored"))
);
let exists = anyhow::Error::new(gunnar_core::Error::RepoExists(
gunnar_core::RepoRef::new("acme", "site").unwrap(),
));
assert!(matches!(
create_error(&exists, &path),
ControlError::AlreadyExists(_)
));
assert!(matches!(
create_error(&anyhow::anyhow!("a store already stands there"), &path),
ControlError::FailedPrecondition(_)
));
}
}
/// A catalog error, mapped to the status code that says **whose fault it is**.
///
/// This used to be `ControlError::Internal(err.to_string())` for every one of
/// its thirty-four call sites, so `gunnar control repo set-default-branch
/// team/project trunk` came back as
///
/// ```text
/// gunnar: Internal error: internal error: not a legal default branch name: "trunk"
/// ```
///
/// An `INTERNAL` says *the server is broken*. What actually happened is that
/// the caller typed a short branch name where a full ref was required, and
/// nothing in that sentence tells them to fix their argument rather than read
/// the server's logs. Every "this repository does not exist", "that name is
/// taken" and "you are not allowed to" arrived the same way.
///
/// `enrolment_error` already classified its own errors for exactly this
/// reason; this is the same job for the rest of the surface, in one function
/// rather than at thirty-four call sites (one implementation). The mapping follows gRPC's
/// own definitions:
///
/// * **INVALID_ARGUMENT** — the request is malformed however the server feels.
/// * **NOT_FOUND** / **ALREADY_EXISTS** — the named thing is or is not there.
/// * **FAILED_PRECONDITION** — well formed, and the server's *state* refuses.
/// * **PERMISSION_DENIED** — authenticated, not allowed.
/// * **UNAVAILABLE** — storage did not answer; retrying may work.
/// * **INTERNAL** — only what is genuinely a bug here, which is the point:
/// an `INTERNAL` an operator sees now means something.
fn internal(err: gunnar_core::Error) -> ControlError {
use gunnar_core::Error as E;
match err {
// The caller's argument.
e @ (E::InvalidPath { .. }
| E::PathEscape { .. }
| E::ReservedPrincipalId(_)
| E::BadBranchName(_)
| E::BadFingerprint(_)
| E::Key(_)
| E::InvalidInput { .. }) => ControlError::InvalidArgument(e.to_string()),
// What is not there.
e @ (E::NoSuchRepo(_)
| E::NoSuchNamespace(_)
| E::NoSuchPrincipal(_)
| E::NoSuchCredential(_)) => ControlError::NotFound(e.to_string()),
// …and what already is.
e @ (E::RepoExists(_)
| E::NamespaceExists(_)
| E::PrincipalExists(_)
| E::NameExists { .. }
| E::KeyExists { .. }
| E::CredentialExists { .. }
| E::InviteExists(_)
| E::TokenExists { .. }
| E::AlreadyBootstrapped { .. }
| E::AlreadyProvisioned { .. }) => ControlError::AlreadyExists(e.to_string()),
// Well formed; the server's state says no. A namespace that still holds
// repositories is the case an operator meets, and `--delete-repos` is
// the answer — which is a precondition, not a bad request.
e @ (E::NamespaceNotEmpty { .. }
| E::Contended { .. }
| E::NotAPerson { .. }
| E::NotAMachine { .. }) => ControlError::FailedPrecondition(e.to_string()),
e @ (E::Denied { .. } | E::CounterRollback { .. } | E::InviteInvalid) => {
ControlError::PermissionDenied(e.to_string())
}
// Storage did not answer. NOT `Internal`: the request may well succeed
// on a retry, and telling a caller their request was wrong when the
// disk was busy sends them to fix the wrong thing.
e @ (E::Store(_) | E::Json(_)) => ControlError::Unavailable(e.to_string()),
// The control store is OFF THE STRIPE while the data set grows: the
// write is refused by name and the caller retries after the growth —
// D7's "refuse", and UNAVAILABLE says exactly that to a generated
// client where INTERNAL would say "look at this server".
e @ E::OffStripe { .. } => ControlError::Unavailable(e.to_string()),
// A box being restored takes no write until the restore is whole: "try again".
e @ E::LogFollows => ControlError::Unavailable(e.to_string()),
// A document that will not parse, and anything unclassified. These are
// what genuinely mean "look at this server". `gunnar_core::Error` is
// `#[non_exhaustive]`, so a variant added later lands here — `INTERNAL`
// is the right default for "gunnar does not know what this is", and it
// errs toward the server's fault rather than toward blaming the caller
// for something nobody has classified.
e => ControlError::Internal(e.to_string()),
}
}
/// **An archive cut's error, classified by whose fault it is** — the same job
/// [`internal`] does for `gunnar_core::Error`, for the one path that reports an
/// `anyhow` chain instead.
///
/// Two classes, and they are told apart by what is *in* the chain rather than
/// by matching on a message:
///
/// * an `io::Error` anywhere in it means the disk or the socket did not answer
/// — `UNAVAILABLE`, because a retry may well work, and a `BrokenPipe` is the
/// *client* having hung up rather than this server being broken;
/// * anything else is a refusal the cut made on purpose — the watermark above a
/// pack a GC has rewritten, a directory with no `objects.pack`, a repository
/// directory holding two stores. Those are `FAILED_PRECONDITION`: the request
/// is well formed and the server's state says no. A full archive
/// (`--from-generation 0`) is the answer to the first, and the message says so
/// in the plugin's own words.
#[cfg(feature = "engine-znippy")]
fn archive_error(err: anyhow::Error) -> ControlError {
let io = err
.chain()
.any(|cause| cause.downcast_ref::<std::io::Error>().is_some());
let message = format!("{err:#}");
if io {
ControlError::Unavailable(message)
} else {
ControlError::FailedPrecondition(message)
}
}
/// **The one storage-fault conversion in this file** (one implementation).
///
/// `GitOps` and `GitServe` answer in `anyhow`, so every read of a repository
/// arrives here. `{:#}` and not `to_string()`: the cause chain is the whole of
/// what an operator has to go on, and `to_string()` prints only the outermost
/// context — *"reading the ref namespace"* with the errno dropped.
///
/// **The status code is unchanged, deliberately.** `Internal` is what this
/// answered before, and `internal_or_denied` twenty lines above classifies
/// `gunnar_core::Error::Store` as `Unavailable` for a reason that applies here
/// too — a storage fault may well succeed on a retry. Changing it is a change
/// to what every gRPC client of `Refs.*` retries on, so it is **recorded here
/// and not taken as a drive-by** of a compile fix.
///
/// ⚠ **A lost compare-and-swap and a broken disk arrive as the same string** on
/// this path. `git-storage-trait` carries [`RefRejection`] for exactly that
/// distinction and `RefRejection::of(&err)` is how it is recovered; nothing in
/// this file needs it yet. Whatever does must use that, and **never** match on
/// the message text.
///
/// [`RefRejection`]: gunnar_store::git_storage_trait::RefRejection
fn store_error(err: anyhow::Error) -> ControlError {
ControlError::Internal(format!("{err:#}"))
}
/// The same, for the one authority on following a symbolic reference.
///
/// [`gunnar_wire::refs`] returns `gunnar_wire::Error` rather than
/// `anyhow::Error`; a symref walk that refuses is a fault in what the store
/// holds, which is the same `Internal` this control plane already reports for
/// [`store_error`], so the two funnel to one variant rather than inventing a
/// second grade of failure for the same event.
fn wire_error(err: gunnar_wire::Error) -> ControlError {
ControlError::Internal(format!("{err:#}"))
}
// ── model translation ───────────────────────────────────────────────────────
/// The one translation of a hash kind into the proto enum. `transport.rs`
/// reuses it for the push attestation rather than writing a second `match`
/// that could drift from this one (one implementation).
pub(crate) fn hash_kind(kind: HashKind) -> gunnar_proto::HashKind {
match kind {
HashKind::Sha1 => gunnar_proto::HashKind::Sha1,
HashKind::Sha256 => gunnar_proto::HashKind::Sha256,
}
}
/// gunnar has **two** visibilities, not three.
///
/// The proto still carries `VISIBILITY_INTERNAL`, and it used to alias
/// `PUBLIC` here — honestly, at the time: with no anonymous path at all,
/// "readable by any authenticated principal" and "public" named the same set.
/// Anonymous read (2026-08-03) broke that equivalence — `Public` now means readable by
/// ANYONE, with no key — and the alias survived the redefinition. What that
/// cost, MEASURED 2026-08-19 through the console: an operator chose
/// "internal" and the repository came out anonymous-readable. Silently
/// WIDENING an access request is the one direction a coercion must never
/// fail, so `INTERNAL` is now refused by name instead of mapped. The day a
/// real internal tier exists in [`Visibility`], this arm becomes its
/// translation; until then the wire value is a claim with no implementation.
fn visibility_in(v: gunnar_proto::Visibility) -> ControlResult<Visibility> {
match v {
gunnar_proto::Visibility::Public => Ok(Visibility::Public),
gunnar_proto::Visibility::Internal => Err(ControlError::InvalidArgument(
"visibility \"internal\" has no implementation: since 2026-08-03, \"public\" means \
readable with no key at all, so \"internal\" (any authenticated principal) \
no longer aliases it and gunnar refuses to guess which you meant. Choose \
\"private\" or \"public\""
.into(),
)),
gunnar_proto::Visibility::Private | gunnar_proto::Visibility::Unspecified => {
Ok(Visibility::Private)
}
}
}
/// The one translation of a stored pull request into the proto shape (one implementation).
fn pull_out(path: &RepoPath, record: gunnar_core::PullRecord) -> PullRequest {
PullRequest {
repo: Some(path.clone()),
number: record.number,
title: record.title,
description: record.description,
source_ref: record.source_ref,
target_ref: record.target_ref,
state: match record.state {
gunnar_core::PullState::Open => PullRequestState::Open,
gunnar_core::PullState::Merged => PullRequestState::Merged,
gunnar_core::PullState::Closed => PullRequestState::Closed,
} as i32,
author: record.author,
created_at_unix_ms: record.created_ms,
updated_at_unix_ms: record.updated_ms,
source_oid: record.source_oid,
merge_oid: record.merge_oid.unwrap_or_default(),
}
}
fn visibility_out(v: Visibility) -> gunnar_proto::Visibility {
match v {
Visibility::Private => gunnar_proto::Visibility::Private,
Visibility::Public => gunnar_proto::Visibility::Public,
}
}
fn principal_kind_in(kind: gunnar_proto::PrincipalKind) -> PrincipalKind {
match kind {
gunnar_proto::PrincipalKind::Machine => PrincipalKind::Machine,
gunnar_proto::PrincipalKind::Deploy => PrincipalKind::Deploy,
gunnar_proto::PrincipalKind::Group => PrincipalKind::Group,
gunnar_proto::PrincipalKind::User | gunnar_proto::PrincipalKind::Unspecified => {
PrincipalKind::User
}
}
}
impl CoreControlPlane {
/// **Whose account the namespace `name` is** (`Namespace.owner_kind`): the
/// kind of the principal whose id it is, else UNSPECIFIED — so a reader
/// that bills per namespace can leave out the estate's own machines by
/// what they ARE, not by a list of their names. A store that cannot be
/// read says UNSPECIFIED (the namespace is still listed).
///
/// A GROUP whose every direct member is a machine is the estate's own too
/// (the `twin` group, the twin's log readers — MEASURED on the mock
/// 2026-09-29, listed as a tenant) and answers MACHINE; a group with anyone
/// else in it, or with nobody, stays a group.
fn owner_kind(&self, name: &str) -> i32 {
let auth = self.stores.auth();
match auth.principal(name) {
Ok(Some(p)) if p.kind == PrincipalKind::Group => {
let members = auth.members_of(name).unwrap_or_default();
let machines = !members.is_empty() && members.iter().all(|m| matches!(auth.principal(m), Ok(Some(q)) if q.kind == PrincipalKind::Machine));
if machines {
gunnar_proto::PrincipalKind::Machine as i32
} else {
principal_kind_out(p.kind) as i32
}
}
Ok(Some(p)) => principal_kind_out(p.kind) as i32,
_ => gunnar_proto::PrincipalKind::Unspecified as i32,
}
}
/// The catalog's namespace onto the wire — one mapping for create, list
/// and the single lookup, so the account's settings and its employee count
/// cannot be on one and missing from another. `employees` is derived (an
/// account with any is an enterprise account; nothing is set).
fn namespace_out(&self, namespace: gunnar_core::Namespace, employees: u32) -> Namespace {
Namespace {
owner_kind: self.owner_kind(&namespace.name),
// A store that cannot be read says SELLER: a customer left out for one tick is given
// its fact on the next, while a fact given to the operator's own namespace stays.
owner_server_grant: self
.stores
.auth()
.holds_server_grant(&namespace.name)
.unwrap_or(true),
name: namespace.name,
description: String::new(),
created_at_unix_ms: 0,
public: namespace.public,
events_public: namespace.events_public,
employees,
}
}
/// How many employees `account` has enrolled.
fn employee_count(&self, account: &str) -> ControlResult<u32> {
Ok(self
.stores
.auth()
.employees_of(account)
.map_err(internal)?
.len() as u32)
}
}
fn principal_kind_out(kind: PrincipalKind) -> gunnar_proto::PrincipalKind {
match kind {
PrincipalKind::User => gunnar_proto::PrincipalKind::User,
PrincipalKind::Machine => gunnar_proto::PrincipalKind::Machine,
PrincipalKind::Deploy => gunnar_proto::PrincipalKind::Deploy,
PrincipalKind::Group => gunnar_proto::PrincipalKind::Group,
// **Unreachable, and mapped honestly rather than plausibly.**
// Every principal that reaches this function came out of the identity
// table, and `AuthTable::add_principal` refuses the anonymous id, so
// there is no anonymous row to list. `gunnar-proto` has no `ANONYMOUS`
// value, and picking `USER` would put a fabricated human in an operator's
// principal list; `UNSPECIFIED` is the value that means "gunnar cannot
// name this", which is the truth. Adding the enum value is a
// `gunnar-proto` change and is recorded there, not invented here.
PrincipalKind::Anonymous => gunnar_proto::PrincipalKind::Unspecified,
}
}
fn level_in(level: GrantLevel) -> ControlResult<Level> {
match level {
GrantLevel::Read => Ok(Level::Read),
GrantLevel::Write => Ok(Level::Write),
GrantLevel::Admin => Ok(Level::Admin),
GrantLevel::Unspecified => Err(ControlError::InvalidArgument(
"grant level unspecified".into(),
)),
}
}
fn level_out(level: Level) -> GrantLevel {
match level {
Level::Read => GrantLevel::Read,
Level::Write => GrantLevel::Write,
Level::Admin => GrantLevel::Admin,
}
}
fn repo_out(repo: &gunnar_core::Repo) -> Repo {
Repo {
path: Some(RepoPath::new(&repo.namespace, &repo.name)),
hash_kind: hash_kind(repo.hash_kind) as i32,
default_branch: repo.default_branch.clone(),
visibility: visibility_out(repo.visibility) as i32,
created_at_unix_ms: repo.created,
description: repo.description.clone(),
encrypted: repo.encrypted,
}
}
/// A stored web credential, as the wire carries it (no key material).
fn web_credential_out(
row: &gunnar_core::WebCredential,
) -> ControlResult<gunnar_proto::WebCredential> {
let credential_id = gunnar_grpc::webauthn::from_b64url(&row.id).ok_or_else(|| {
ControlError::Internal(format!(
"a stored web credential has an id that is not base64url: {}",
row.id
))
})?;
Ok(gunnar_proto::WebCredential {
credential_id,
principal_id: row.principal_id.clone(),
label: row.label.clone(),
created_at_unix_ms: row.created,
last_used_unix_ms: row.last_used,
sign_count: row.sign_count,
})
}
fn principal_out(principal: &gunnar_core::Principal) -> Principal {
Principal {
id: principal.id.clone(),
name: principal.name.clone(),
kind: principal_kind_out(principal.kind) as i32,
// The table's stamp. `0` for a row from before the field existed,
// which is exactly what this reported for every row until then.
created_at_unix_ms: principal.created,
}
}
/// Clans on the identity table (`gunnar_grpc::clans::ClanPlane`, 2026-10-05).
mod clan_plane;
fn key_out(key: &gunnar_core::PublicKey) -> PublicKey {
PublicKey {
fingerprint: key.fingerprint_sha256.clone(),
algo: key.algo.clone(),
principal_id: key.principal_id.clone(),
// The OpenSSH comment field is part of the stored line; gunnar-core
// normalises the line and does not split it out.
comment: String::new(),
added_at_unix_ms: key.added,
last_used_unix_ms: key.last_used,
expires_at_unix_ms: key.expires,
}
}
/// A grant, as `<namespace, repo>` strings.
///
/// A [`Scope::Server`] grant has neither, and is carried as two empty strings.
/// See the module docs: that is a proto gap, not an invention here.
fn grant_out(grant: &Grant) -> gunnar_proto::Grant {
let (namespace, repo) = match &grant.scope {
Scope::Server => (String::new(), String::new()),
Scope::Namespace(name) => (name.clone(), String::new()),
Scope::Repo(path) => (path.namespace().to_owned(), path.name().to_owned()),
};
gunnar_proto::Grant {
principal_id: grant.principal.clone(),
namespace,
repo,
level: level_out(grant.level) as i32,
granted_at_unix_ms: 0,
}
}
fn scope_in(namespace: &str, repo: &str) -> ControlResult<Scope> {
match (namespace.is_empty(), repo.is_empty()) {
(true, true) => Ok(Scope::Server),
(false, true) => Ok(Scope::Namespace(namespace.to_owned())),
(false, false) => RepoRef::new(namespace, repo)
.map(Scope::Repo)
.map_err(|err| ControlError::InvalidArgument(err.to_string())),
(true, false) => Err(ControlError::InvalidArgument(
"a repository-scoped grant must name a namespace (multi-tenancy has \
no ungrouped repos)"
.into(),
)),
}
}
/// The contract's object kind, in the proto's vocabulary.
///
/// [`ObjType`] also has `OfsDelta` and `RefDelta`, and neither can appear here:
/// [`GitServe::read`] and [`GitServe::header`] answer with the **resolved**
/// type by contract — that is the difference between them and
/// `GitOps::get`, which hands back the pack entry delta and all. A delta
/// reaching this function is a backend that broke its own promise, so it is
/// named as one rather than folded into `Unspecified`, which a client would
/// read as "an object of a kind this server is too old to name".
fn object_kind_out(kind: ObjType) -> ControlResult<gunnar_proto::ObjectKind> {
match kind {
ObjType::Blob => Ok(gunnar_proto::ObjectKind::Blob),
ObjType::Tree => Ok(gunnar_proto::ObjectKind::Tree),
ObjType::Commit => Ok(gunnar_proto::ObjectKind::Commit),
ObjType::Tag => Ok(gunnar_proto::ObjectKind::Tag),
other => Err(ControlError::Internal(format!(
"the store answered a read with `{}`, which is a delta; GitServe::read is \
documented to return the resolved type",
other.as_str()
))),
}
}
/// Follow an annotated tag down to the object it names.
///
/// A bounded loop, not a recursion: a tag chain is legal git and a cycle is
/// not, but a cycle is also cheap to write and would otherwise hang a control
/// plane that has no timeout.
///
/// What object id `segments` names under `tree` — a blob's id, a subtree's id,
/// or `None` when the path names nothing there.
///
/// The log's path filter rests on this: a commit changed a path exactly when
/// this answer differs from its first parent's. A path that continues THROUGH
/// a non-directory names nothing, the same rule the wire's tree walk applies.
fn oid_at_path(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
tree: gix_hash::ObjectId,
segments: &[&[u8]],
) -> ControlResult<Option<gix_hash::ObjectId>> {
let mut current = tree;
let last = segments.len().saturating_sub(1);
for (index, segment) in segments.iter().enumerate() {
let Some((kind, bytes)) = git.read(current.as_bytes()).map_err(store_error)? else {
return Ok(None);
};
if kind != ObjType::Tree {
return Ok(None);
}
let Some(entry) = object::tree_entry_named(&bytes, hash_kind, segment)
.map_err(|err| ControlError::Internal(format!("decoding tree {current}: {err}")))?
else {
return Ok(None);
};
if index == last {
return Ok(Some(entry.oid));
}
if entry.kind != object::EntryKind::Tree {
return Ok(None);
}
current = entry.oid;
}
Ok(Some(current))
}
/// `hash_kind` is a parameter and never a default: it comes from
/// [`OpenRepo::hash_kind`], which comes from the catalog entry, and answering
/// `sha1` for a sha256 repository does not error — it decodes garbage.
/// Where a `Refs.Log` walk begins: one start (a ref name, an oid, or empty
/// for `HEAD`), or every branch and tag tip plus `HEAD`.
#[derive(Clone, Copy)]
enum LogStart<'a> {
One(&'a str),
AllRefs,
}
/// One `Refs.Log` start resolved to an oid: an oid as given, otherwise a ref
/// name (empty means `HEAD`) resolved against one snapshot.
fn resolve_log_start(
git: &dyn GitServe,
hash: gix_hash::Kind,
path: &RepoPath,
start: &str,
) -> ControlResult<gix_hash::ObjectId> {
let start = if start.is_empty() {
gunnar_store::HEAD
} else {
start
};
let head = match gix_hash::ObjectId::from_hex(start.as_bytes()) {
Ok(id) => id,
Err(_) => {
gunnar_store::refname::validate(start)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
// `HEAD` is not in the snapshot (the `GitServe` HEAD ruling) and it is this verb's
// own default, so it is asked through the accessor pair. Both
// branches resolve against the same snapshot.
let resolved = if start == gunnar_store::HEAD {
snapshot.head(git).map_err(wire_error)?
} else {
snapshot.resolve(start).map_err(wire_error)?
};
resolved.ok_or_else(|| ControlError::NotFound(format!("{start} in {path}")))?
}
};
if head.kind() != hash {
return Err(ControlError::InvalidArgument(format!(
"{start} is a {} oid; this repository is {hash}",
head.kind()
)));
}
Ok(head)
}
impl CoreControlPlane {
fn log_walk(
&self,
path: &RepoPath,
start: LogStart<'_>,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
use std::collections::{BinaryHeap, HashMap, HashSet};
let (_, store) = self.open(path)?;
let git = store.serve();
let hash = store.hash_kind();
// The path filter, refused loudly rather than normalised: a filter
// that silently ate `..` would answer for a different path than the
// caller named.
let filter: Option<Vec<&[u8]>> = if touching.is_empty() {
None
} else {
let mut segments = Vec::new();
for segment in touching.split('/').filter(|s| !s.is_empty()) {
if segment == "." || segment == ".." {
return Err(ControlError::InvalidArgument(format!(
"the log path filter cannot contain {segment:?}"
)));
}
segments.push(segment.as_bytes());
}
(!segments.is_empty()).then_some(segments)
};
let heads: Vec<gix_hash::ObjectId> = match start {
LogStart::One(start) => vec![resolve_log_start(git, hash, path, start)?],
LogStart::AllRefs => {
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let mut heads = Vec::new();
if let Some(head) = snapshot.head(git).map_err(wire_error)? {
heads.push(head);
}
for row in snapshot.rows().iter().filter(|r| {
r.name.starts_with("refs/heads/") || r.name.starts_with("refs/tags/")
}) {
if let Some(id) = snapshot.resolve_row(row).map_err(wire_error)? {
heads.push(id);
}
}
heads
}
};
// Newest first, by EACH COMMIT'S OWN committer time — so tips of
// several branches interleave the way `git log --date-order --all`
// does, and a page cut by `limit` is the newest `limit` commits.
// Each commit is decoded once, when it is queued, and kept until it
// is popped.
let limit = if limit == 0 { 50 } else { limit } as usize;
let mut heap: BinaryHeap<(i64, gix_hash::ObjectId)> = BinaryHeap::new();
let mut seen: HashSet<gix_hash::ObjectId> = HashSet::new();
let mut queued: HashMap<gix_hash::ObjectId, object::Commit> = HashMap::new();
let mut enqueue = |id: gix_hash::ObjectId,
heap: &mut BinaryHeap<(i64, gix_hash::ObjectId)>,
queued: &mut HashMap<gix_hash::ObjectId, object::Commit>|
-> ControlResult<()> {
if !seen.insert(id) {
return Ok(());
}
// A shallow edge (not in this store) or a tip that peels to a
// non-commit ends that line quietly, as it always did.
let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
return Ok(());
};
if kind != ObjType::Commit {
return Ok(());
}
let commit = object::commit(&bytes, hash)
.map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}")))?;
heap.push((commit.committer.seconds, id));
queued.insert(id, commit);
Ok(())
};
for head in heads {
let head = peel(git, hash, head)?;
enqueue(head, &mut heap, &mut queued)?;
}
let mut out = Vec::new();
// What a FILTERED walk may visit before giving up. A path nothing ever
// touched must not cost a full traversal of a quarter-million commits
// per page; a page that comes back short because of this bound is a
// short page, not the end of history — the proto says so.
const MAX_FILTERED_SCAN: usize = 20_000;
let mut scanned = 0usize;
// commit → its tree, remembered so a parent decoded for the filter is
// not decoded again when the walk reaches it — and vice versa.
let mut tree_of: HashMap<gix_hash::ObjectId, gix_hash::ObjectId> = HashMap::new();
while let Some((_, id)) = heap.pop() {
if out.len() >= limit {
break;
}
let Some(commit) = queued.remove(&id) else {
continue;
};
let object::Commit {
tree,
parents,
author,
committer,
message,
} = commit;
// ── the path filter: did THIS commit change what the path names? ─
//
// Judged against the FIRST parent, which is what `git log -- path`
// shows for ordinary history: "what did landing this commit change
// here". Identical ids mean identical content all the way down, so
// the question costs two short descents, never a blob.
if let Some(segments) = &filter {
scanned += 1;
if scanned > MAX_FILTERED_SCAN {
break;
}
tree_of.insert(id, tree);
let current = oid_at_path(git, hash, tree, segments)?;
let previous = match parents.first() {
None => None,
Some(parent) => {
let parent_tree = match tree_of.get(parent) {
Some(known) => Some(*known),
None => match git.read(parent.as_bytes()).map_err(store_error)? {
Some((ObjType::Commit, parent_bytes)) => {
let decoded =
object::commit(&parent_bytes, hash).map_err(|err| {
ControlError::Internal(format!(
"decoding commit {parent}: {err}"
))
})?;
tree_of.insert(*parent, decoded.tree);
Some(decoded.tree)
}
// A shallow edge: the parent is not in this
// store. Comparing against nothing keeps the
// commit visible rather than silently dropped.
_ => None,
},
};
match parent_tree {
Some(parent_tree) => oid_at_path(git, hash, parent_tree, segments)?,
None => None,
}
}
};
if current == previous {
// Untouched here — keep walking, show nothing.
for parent in parents {
enqueue(parent, &mut heap, &mut queued)?;
}
continue;
}
}
out.push(Commit {
oid: id.to_string(),
parent_oids: parents.iter().map(|p| p.to_string()).collect(),
tree_oid: tree.to_string(),
author: Some(signature_out(
&author.name,
&author.email,
author.seconds,
author.offset_seconds,
)),
committer: Some(signature_out(
&committer.name,
&committer.email,
committer.seconds,
committer.offset_seconds,
)),
message: String::from_utf8_lossy(&message).into_owned(),
});
for parent in parents {
enqueue(parent, &mut heap, &mut queued)?;
}
}
Ok(out)
}
}
fn peel(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
mut id: gix_hash::ObjectId,
) -> ControlResult<gix_hash::ObjectId> {
for _ in 0..16 {
let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
return Ok(id);
};
if kind != ObjType::Tag {
return Ok(id);
}
id = object::tag_target(&bytes, hash_kind)
.map_err(|err| ControlError::Internal(format!("decoding tag {id}: {err}")))?;
}
Err(ControlError::FailedPrecondition(format!(
"the tag chain at {id} is longer than 16 links, or cyclic"
)))
}
/// One [`RefRow`] as the proto's `RefTarget`, resolved against `snapshot` — the
/// **same** snapshot the row came out of, so an answer cannot name an oid for a
/// reference that had already moved. See [`gunnar_wire::refs`].
fn ref_target_out(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
snapshot: &RefSnapshot,
row: &RefRow,
want_peeled: bool,
) -> ControlResult<RefTarget> {
let (oid, symbolic) = match (&row.oid, &row.symref_target) {
// A symbolic reference is resolved through the snapshot, whatever else
// it carries: the row's own `oid` on a symref is the store's cache of
// the same walk, and one authority is enough.
(_, Some(inner)) => (snapshot.resolve(inner).map_err(wire_error)?, inner.clone()),
(Some(raw), None) => (
Some(gunnar_wire::refs::oid(raw).map_err(wire_error)?),
String::new(),
),
(None, None) => (None, String::new()),
};
let peeled = match (want_peeled, oid) {
// `row.peeled` is not used even when the store filled it in: `peel`
// follows the whole chain and a stored peel is one hop, so taking the
// cheaper answer where one exists would make two references with the
// same tag chain answer differently.
(true, Some(id)) => {
let peeled = peel(git, hash_kind, id)?;
if peeled == id {
String::new()
} else {
peeled.to_string()
}
}
_ => String::new(),
};
Ok(RefTarget {
ref_name: row.name.clone(),
oid: oid.map(|id| id.to_string()).unwrap_or_default(),
symbolic_target: symbolic,
peeled_oid: peeled,
})
}
/// A commit's author or committer line, as the gRPC message spells it.
///
/// **Bytes and numbers, not `gix_actor::SignatureRef`.** That was the parameter
/// type until 2026-08-10, and it was the whole reason `gix-actor` was a direct
/// dependency of this crate: one private four-line function, in the manifest
/// beside `gix-pack`, where the next reader had to work out whether the server
/// was *using* an actor parser or merely naming one. It was merely naming one.
/// `gix-object` still arrives here — it is what decodes the commit these fields
/// come out of — and it brings `gix-actor` with it; what is gone is gunnar
/// asserting a need it does not have.
///
/// `from_utf8_lossy` matches what `BStr`'s own `Display` did before: a name in
/// a commit header is whatever the committing client wrote, a gRPC string field
/// is UTF-8, and this is a *report* about history rather than anything looked
/// up by name.
fn signature_out(name: &[u8], email: &[u8], seconds: i64, tz_offset_seconds: i32) -> Signature {
Signature {
name: String::from_utf8_lossy(name).into_owned(),
email: String::from_utf8_lossy(email).into_owned(),
time_unix_seconds: seconds,
tz_offset_minutes: tz_offset_seconds / 60,
}
}
// ───────────────────────────────────────────────────────────────────────────
/// **First boot, delegated straight to `gunnar_core::AuthTable`.**
///
/// Nothing is decided here. `bootstrap_state` reads the one predicate the
/// writer also reads, and `enroll_first_identity` is a call into the one
/// compare-and-swap that can produce a `BootstrapRecord`. A second opinion in
/// this adapter — "the table looks empty, so the window is open" — would be a
/// status that could disagree with the gate, and the permissive one would win
/// silently (one implementation).
impl gunnar_grpc::FirstBoot for CoreControlPlane {
fn bootstrap_state(&self) -> ControlResult<gunnar_grpc::BootstrapState> {
let auth = self.stores.auth();
// Every one of these is a `Result`. A read fault becomes UNAVAILABLE
// and never a window that is reported as open or as shut on a guess.
let window_open = auth.bootstrap_window_open().map_err(unavailable)?;
let record = auth.bootstrap_record().map_err(unavailable)?;
let (principals, keys, _grants) = auth.counts().map_err(unavailable)?;
Ok(gunnar_grpc::BootstrapState {
window_open,
bootstrapped: record.is_some(),
bootstrapped_at_unix_ms: record.as_ref().map(|r| r.at).unwrap_or_default(),
bootstrapped_principal: record
.as_ref()
.map(|r| r.principal_id.clone())
.unwrap_or_default(),
bootstrapped_fingerprint: record
.as_ref()
.map(|r| r.fingerprint.clone())
.unwrap_or_default(),
principals: principals as u64,
keys: keys as u64,
// The boot window (`crate::readiness`). Answered here rather than
// on `Admin.Health` because this is the one call a server nobody
// holds a key for can answer, and a boot script is exactly that
// caller.
// TWO facts, and the second one is new. The boot window is this
// process's start-up; the listener is a socket. A twin closes no
// window and binds no reachable git port, so before the listener
// was consulted this field answered `true` for a box whose :2222
// was answering ECONNREFUSED — measured on 10.13.8.101, 2026-09-20.
serving_git: crate::readiness::is_serving_git(self.git_listener),
readiness: crate::readiness::detail_with(self.git_listener),
build: gunnar_common::BUILD_REF.to_owned(),
role: if self.twin_role || self.stores.is_twin() {
"twin"
} else {
"primary"
}
.to_owned(),
// A RECHARGED main (the installer's re-admit) or a HELD one (the boot
// check, R-7b) says what it awaits (the restore lane reads it).
recharge_from: awaited_twin(&self.stores),
restoring: crate::twin::is_restoring(self.stores.root()),
log_head_seq: self.stores.txlog().head_seq(),
// What a restoring box holds of its next entry: where the next stream resumes it.
restore_partial: if crate::twin::is_restoring(self.stores.root()) {
crate::twin::intake::partial(self.stores.root())
} else {
None
},
// restore-after-snapshot: this box stages a twin's base across streams, and says how
// much of one it holds while it is restoring.
restore_snapshot_resumable: true,
restore_snapshot: if crate::twin::is_restoring(self.stores.root()) {
crate::twin::snapshot::held_snapshot(self.stores.root())
.map(|h| (h.id, h.items, h.bytes, h.whole))
} else {
None
},
})
}
fn enroll_first_identity(
&self,
name: &str,
openssh_public_key: &str,
) -> ControlResult<gunnar_grpc::Enrolled> {
// R-7b: a box the boot check found with its data lost (a twin can give it
// back) takes no first admin while it is held.
#[cfg(feature = "replication")]
if let Some(twin) = crate::repair::held(&self.stores) {
return Err(ControlError::FailedPrecondition(
crate::repair::held_refusal(&twin),
));
}
let enrolled = self
.stores
.auth()
.enroll_first_user(name, openssh_public_key)
.map_err(enrolment_error)?;
// The account namespace, after the door is shut. Deliberately NOT
// inside the identity CAS: it lives in the catalog, a different
// document, and one write cannot span both. The ordering is what makes
// that safe — the security boundary (the record) has already landed, so
// the worst a failure here can do is leave a gunnar-admin whose account
// namespace has to be created by hand, never a second open window.
//
// Idempotent: `AlreadyExists` on a namespace this principal already
// intrinsically owns is not a failure of enrolment.
if let Err(err) = self
.stores
.catalog()
.create_namespace(&enrolled.record.account)
{
tracing::warn!(
account = %enrolled.record.account,
%err,
"the first user was enrolled and the enrolment window is closed, but its account \
namespace could not be created. The user is admin of it by rule and can create \
it themselves with Repos.CreateNamespace."
);
}
// ── the bootstrap admin's client certificate ────────────────────────
//
// Minted AFTER the CAS, so the security boundary is already shut when
// this runs and a failure here cannot reopen it — the same ordering
// argument as the namespace above. Server-generated key, deliberately:
// there is no authenticated channel yet to carry a CSR, and this is the
// one exchange in the server's life where handing a private key over
// the wire is the honest trade. The server retains no copy of the key.
let (client_cert_pem, client_key_pem, issuer_cert_pem) = match self.stores.pki() {
Ok(pki) => match pki.issuer() {
Ok(Some(issuer)) => {
match pki.issue_with_server_key(
&enrolled.principal.id,
gunnar_core::DEFAULT_LEAF_DAYS,
) {
Ok(issued) => {
if let Err(err) = self.stores.auth().register_certificate(
&enrolled.principal.id,
&issued.cert_der,
None,
) {
tracing::warn!(
principal = %enrolled.principal.id,
%err,
"the bootstrap certificate was minted but could not be \
registered; it is returned anyway and gunnar-admin can \
register it with `gunnar key add`"
);
}
(
issued.cert_pem,
issued.key_pem.unwrap_or_default(),
issuer.cert_pem,
)
}
Err(err) => {
tracing::warn!(
principal = %enrolled.principal.id,
%err,
"no bootstrap certificate: the issuer refused to mint"
);
(String::new(), String::new(), String::new())
}
}
}
Ok(None) => (String::new(), String::new(), String::new()),
Err(err) => {
tracing::warn!(%err, "no bootstrap certificate: the PKI document could not \
be read");
(String::new(), String::new(), String::new())
}
},
Err(_) => (String::new(), String::new(), String::new()),
};
Ok(gunnar_grpc::Enrolled {
principal: principal_out(&enrolled.principal),
key: key_out(&enrolled.key),
account: enrolled.record.account,
bootstrapped_at_unix_ms: enrolled.record.at,
client_cert_pem,
client_key_pem,
issuer_cert_pem,
})
}
}
/// A read fault on the identity table. **Never a denial and never a state.**
fn unavailable(err: gunnar_core::Error) -> ControlError {
ControlError::Unavailable(format!("the identity table could not be read: {err}"))
}
/// The refusals of the enrolment window, mapped so a caller can tell them apart
/// from a malformed request.
/// **The twin this box awaits** (`Bootstrap.Status.recharge_from`): the installer's
/// re-admit ([`crate::pairing::recharge_awaited`]), else the boot check's hold
/// ([`crate::repair::held`]); empty on every other box.
fn awaited_twin(stores: &ServerStores) -> String {
#[cfg(feature = "wg-client")]
if let Some(twin) = crate::pairing::recharge_awaited(stores) {
return twin;
}
#[cfg(feature = "replication")]
if let Some(twin) = crate::repair::held(stores) {
return twin;
}
let _ = stores;
String::new()
}
fn enrolment_error(err: gunnar_core::Error) -> ControlError {
match err {
// The door is shut. FAILED_PRECONDITION, because the request was well
// formed and the *server's state* is what refuses it.
e @ (gunnar_core::Error::AlreadyBootstrapped { .. }
| gunnar_core::Error::AlreadyProvisioned { .. }) => {
ControlError::FailedPrecondition(e.to_string())
}
// A bad name or a key that is not an authorized_keys line.
e @ (gunnar_core::Error::InvalidPath { .. }
| gunnar_core::Error::ReservedPrincipalId(_)
| gunnar_core::Error::Key(_)) => ControlError::InvalidArgument(e.to_string()),
// Storage. Not "the door is shut", which is what a caller would read a
// FAILED_PRECONDITION as, and not OK, which would be the disaster.
other => ControlError::Unavailable(other.to_string()),
}
}
impl CoreControlPlane {
/// ★ **The estate's operator, made on its first claim** — the one writer
/// behind `Accounts.ClaimOperator` and behind the primary's own claim when
/// the installer feeds `operator_name` ([`crate::fed_principals::install`]).
///
/// `fed` is the installer's name for the operator and wins; `asked` is the
/// caller's, used only when nothing was fed (`gunnar_grpc::seam::operator_name`).
///
/// * A PERSON already under that id or name is the answer, unchanged but
/// for two upserts that make it what an operator is: its account
/// namespace exists, and it is gunnar-admin.
/// * Anything else under it — a machine, a group — is refused by name: an
/// estate's parts are never its operator.
/// * Otherwise the principal is created, KEYLESS (the console's credential
/// for the person is its first key, which owes nobody an attestation; the
/// person's own SSH keys come after and are vouched for by it), with its
/// account namespace — and **a namespace of that name that already exists
/// with no owner is ADOPTED** rather than refused. MEASURED on test2
/// 2026-10-04: a first push with the installer's admin key
/// (`rickard/t2-scratch`) conjured `rickard`, and `AddPrincipal` refused
/// the operator's own account from then on — the first admin's passkey
/// bound to nothing. Adoption is safe exactly here: the namespace has no
/// principal of its id (this call is creating it), only a server Admin can
/// ask, and that Admin could already delete the namespace or grant it.
pub fn claim_operator_named(
&self,
fed: Option<&str>,
asked: &str,
) -> ControlResult<gunnar_grpc::OperatorClaim> {
let (name, named_by) = gunnar_grpc::seam::operator_name(fed, asked)?;
let auth = self.stores.auth();
let held = match auth.principal(name).map_err(internal)? {
Some(p) => Some(p),
None => auth.principal_by_name(name).map_err(internal)?,
};
if let Some(p) = held {
if p.kind != PrincipalKind::User {
return Err(ControlError::FailedPrecondition(format!(
"the operator's name {name:?} is held by principal {:?}, which is not a person \
({:?}); an estate's machines are never its operator",
p.id, p.kind
)));
}
let adopted_namespace = false;
if self
.stores
.catalog()
.namespace(&p.id)
.map_err(internal)?
.is_none()
{
self.stores
.catalog()
.create_namespace(&p.id)
.map_err(internal)?;
}
self.grant(&p.id, "", "", GrantLevel::Admin)?;
return Ok(gunnar_grpc::OperatorClaim {
principal: principal_out(&p),
created: false,
adopted_namespace,
named_by,
});
}
let principal =
gunnar_core::Principal::new(name, name, PrincipalKind::User).map_err(internal)?;
let created = auth.add_principal(principal).map_err(internal)?;
let adopted_namespace = match self.stores.catalog().create_namespace(&created.id) {
Ok(_) => false,
Err(gunnar_core::Error::NamespaceExists(_)) => true,
Err(err) => {
let _ = auth.remove_principal(&created.id);
return Err(internal(err));
}
};
if let Err(err) = self.grant(&created.id, "", "", GrantLevel::Admin) {
// Not left half-made: an operator that is not the estate's admin is
// not the one that was asked for, and the next claim makes it whole.
let _ = auth.remove_principal(&created.id);
return Err(err);
}
self.stores.mirror_identity();
tracing::warn!(
target: "gunnar::audit",
event = "operator.claimed",
operator = %created.id,
adopted_namespace,
named_by,
"the estate's operator was created: a person, keyless, gunnar-admin, with its account \
namespace{}",
if adopted_namespace {
" — the namespace already existed with no owner (a first push conjured it) and is \
now the operator's account"
} else {
""
}
);
Ok(gunnar_grpc::OperatorClaim {
principal: principal_out(&created),
created: true,
adopted_namespace,
named_by,
})
}
}
/// The owner's enterprise list on the real stores (`gunnar_grpc::enterprise::EnterprisePlane`).
mod enterprise_plane;
impl ControlPlane for CoreControlPlane {
fn enterprise(&self) -> Option<&dyn gunnar_grpc::enterprise::EnterprisePlane> {
Some(self)
}
// ── namespaces ─────────────────────────────────────────────────────────
fn create_namespace(&self, name: &str, _description: &str) -> ControlResult<Namespace> {
// An EMPLOYEE's id is never a namespace: a namespace named after a
// principal makes that principal its admin (`owns_account`), and an
// employee is admin of nothing (`AuthTable::enroll_employee`).
if self
.stores
.auth()
.employer_of(name)
.map_err(internal)?
.is_some()
{
return Err(ControlError::AlreadyExists(format!(
"{name} is an employee's id, and an employee holds no account of its own"
)));
}
let namespace = self
.stores
.catalog()
.create_namespace(name)
.map_err(internal)?;
Ok(self.namespace_out(namespace, 0))
}
fn list_namespaces(&self) -> ControlResult<Vec<Namespace>> {
let mut employees: std::collections::BTreeMap<String, u32> = Default::default();
for principal in self.stores.auth().principals().map_err(internal)? {
if let Some(account) = self
.stores
.auth()
.employer_of(&principal.id)
.map_err(internal)?
{
*employees.entry(account).or_default() += 1;
}
}
Ok(self
.stores
.catalog()
.namespaces()
.map_err(internal)?
.into_iter()
.map(|ns| {
let count = employees.get(&ns.name).copied().unwrap_or(0);
self.namespace_out(ns, count)
})
.collect())
}
/// One catalog probe, not the listing the trait's default walks: this is
/// asked by the account settings and the outsider's view of a namespace.
fn namespace(&self, name: &str) -> ControlResult<Option<Namespace>> {
let Some(ns) = self.stores.catalog().namespace(name).map_err(internal)? else {
return Ok(None);
};
let count = self.employee_count(name)?;
Ok(Some(self.namespace_out(ns, count)))
}
fn namespace_for_outsider(
&self,
principal_id: &str,
name: &str,
) -> ControlResult<Option<Namespace>> {
// The gate, both directions, before the setting — the same walk
// `AuthTable::may` makes (an ancestor's gate closes a nested account).
let auth = self.stores.auth();
if auth
.gate_covering(principal_id)
.map_err(internal)?
.is_some()
|| auth.gate_covering(name).map_err(internal)?.is_some()
{
return Ok(None);
}
self.namespace(name)
}
fn set_namespace_settings(
&self,
name: &str,
public: Option<bool>,
events_public: Option<bool>,
) -> ControlResult<Namespace> {
let catalog = self.stores.catalog();
let mut ns = catalog
.namespace(name)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("namespace {name}")))?;
if let Some(public) = public {
ns = catalog
.set_namespace_public(name, public)
.map_err(internal)?;
}
if let Some(events_public) = events_public {
ns = catalog
.set_namespace_events_public(name, events_public)
.map_err(internal)?;
}
let count = self.employee_count(name)?;
Ok(self.namespace_out(ns, count))
}
fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32> {
// A CLAN's namespace goes with the clan (`Clans.Dissolve`), never on
// its own: a clan left without its namespace would be a name nobody
// could make anything in, and the namespace's repositories are the
// clan's to dissolve, not one admin grant's to delete.
if self.stores.auth().is_clan(name).map_err(internal)? {
return Err(ControlError::FailedPrecondition(format!(
"namespace {name} is clan {name}'s: it goes when the clan's admin dissolves the clan"
)));
}
let gone = self
.stores
.catalog()
.delete_namespace(name, delete_repos)
.map_err(internal)?;
// **The flow half of deleting a tenant.** The activity ledger keeps
// four hundred days per namespace INCLUDING which key fingerprints
// pushed, and a tenant that is gone must not leave a record of who
// worked on it. Deliberately after the catalog delete: if that refused
// — a non-empty namespace without `delete_repos` — nothing has been
// deleted and the series must stay.
//
// A failure here is logged and not returned: the namespace IS gone, and
// reporting the delete as failed would send an operator to re-run it
// against a name that no longer exists.
if let Err(err) = self.stores.flow().forget(name) {
tracing::warn!(
target: "gunnar::store",
event = "flow.forget_failed",
namespace = %name,
%err,
"the namespace was deleted but its activity series could not be removed"
);
}
Ok(gone.len() as u32)
}
// ── repositories ───────────────────────────────────────────────────────
fn create_repo(&self, spec: ProtoNewRepo) -> ControlResult<Repo> {
let path = Self::repo_ref(&spec.path)?;
// **The disk, before the row.** A repository created on a full disk is
// a catalog row whose store cannot be materialised, which is the
// "listed but git says it does not exist" shape `ServerStores::create_repo`
// exists to prevent — refused here, by the gate the front doors share,
// with the line a refused push prints.
if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
return Err(ControlError::ResourceExhausted(line));
}
let mut new = gunnar_core::NewRepo::new(path).visibility(visibility_in(spec.visibility)?);
if spec.hash_kind != gunnar_proto::HashKind::Unspecified {
new = new.hash_kind(match spec.hash_kind {
gunnar_proto::HashKind::Sha1 => HashKind::Sha1,
_ => HashKind::Sha256,
});
}
if !spec.default_branch.is_empty() {
new = new.default_branch(spec.default_branch.clone());
}
// The catalog entry and the on-disk repository are created together
// and transactionally — `ServerStores::create_repo` looks at the disk
// first (a store already standing at the name is refused BY NAME
// unless `adopt`), writes the row, makes the store, and rolls the row
// back if the store fails. Creating one without the other is the
// shape that produces "it is listed but git says it does not exist";
// adopting a stranger's directory is the shape that resurrected a
// deleted repository on 2026-08-30.
let repo = self
.stores
.create_repo(new, spec.adopt)
.map_err(|err| create_error(&err, &spec.path))?;
Ok(repo_out(&repo))
}
fn write_first_commit(&self, path: &RepoPath, readme: &str) -> ControlResult<String> {
let (repo, store) = self.open(path)?;
let kind = match repo.hash_kind {
HashKind::Sha1 => gix_hash::Kind::Sha1,
_ => gix_hash::Kind::Sha256,
};
let when = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0);
let (pack, commit) = crate::first_commit::first_commit_pack(
kind,
"README.md",
readme.as_bytes(),
"Welcome",
when,
)
.map_err(ControlError::InvalidArgument)?;
store
.serve()
.put(
&pack,
&[gunnar_store::git_storage_trait::RefUpdate::set(
repo.default_branch.clone(),
commit.clone(),
)],
)
.map_err(|err| {
ControlError::Internal(format!("writing the first commit of {path}: {err:#}"))
})?;
Ok(commit)
}
/// The front doors' plan chain (`ServerStores::plan_gates`), asked as a
/// push-to-create asks it: the verdict on a write (a suspended or retained
/// plan: `FailedPrecondition`), then the cap (`ResourceExhausted`). A gate
/// that does not bind machines is not asked for one.
fn create_room(&self, namespace: &str, machine: bool) -> ControlResult<()> {
use gunnar_ssh::{PlanAction, PlanVerdict};
for gate in self
.stores
.plan_gates()
.iter()
.filter(|gate| !machine || gate.applies_to_machine())
{
if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
return Err(ControlError::FailedPrecondition(line));
}
if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
return Err(ControlError::ResourceExhausted(line));
}
}
Ok(())
}
fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.as_ref()
.map(repo_out)
.ok_or_else(|| ControlError::NotFound(format!("repository {path}")))
}
fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>> {
Ok(self
.stores
.catalog()
.list(namespace)
.map_err(internal)?
.iter()
.map(repo_out)
.collect())
}
/// **Delete: a dated tombstone by default, a reclaim when asked.**
///
/// Two facts had to be held at once here, and they pull in opposite
/// directions:
///
/// * **The reclaim**: `repo delete` must reclaim the disk. Before it, the
/// catalog row went and the objects stayed — 25 GB of a kernel push left
/// on the appliance with its indexer still running, invisible because
/// nothing listed it.
/// * **Measured**: the bytes are gone in ~2 seconds (181 GiB to 33 MiB
/// measured) and a mirror does not help, because the delete replicates.
/// The failure a paying customer is most likely to cause is deleting
/// their own repository.
///
/// A tombstone answers both, and only because it is **dated and listed**:
/// the bytes are held on purpose with a deadline written at the instant of
/// the delete, [`ServerStores::sweep_expired_tombstones`] reclaims them
/// when it passes, and `Repos.ListDeleted` / `Admin.StorageInfo` report
/// them so they are never the invisible 25 GB again. **They still count
/// against the account** — see [`Self::storage_info`].
fn delete_repo(
&self,
path: &RepoPath,
options: gunnar_grpc::seam::DeleteOptions,
) -> ControlResult<gunnar_grpc::seam::DeleteOutcome> {
let repo_ref = Self::repo_ref(path)?;
// The row is read BEFORE it is removed: the tombstone records what the
// repository was, and an undelete that guessed the hash kind would
// reopen the resurrection the reclaim rules closed from the other side.
let was = self.stores.catalog().get(&repo_ref).map_err(internal)?;
// **An ORPHAN is a delete to finish, not a not-found** (2026-08-30). A
// store on disk with no row and no tombstone is what an old
// catalog-only delete left, and on the appliance 2026-08-30 it was ~25-39
// GB at `alice/linux` that `delete` answered with `NoSuchRepo` after
// the access check had already refused everybody. `--purge` is
// honoured — the bytes are what the operator asked for and the row is
// already gone — and a SOFT delete is refused by name, because a
// tombstone with no row promises an undelete the server cannot honour.
let orphan = was.is_none() && self.stores.is_orphan(&repo_ref).unwrap_or(false);
if orphan && !options.purge {
return Err(ControlError::FailedPrecondition(
crate::stores::ServerStores::orphan_soft_delete_refusal(&repo_ref),
));
}
// A name that is not in the catalog but is TOMBSTONED (a previous
// delete could not reclaim its store) is a delete to retry, not a
// not-found: the row is already gone, the reclaim is what is owed.
match self.stores.catalog().delete(&repo_ref) {
Ok(_) => {}
Err(gunnar_core::Error::NoSuchRepo(_))
if orphan || self.stores.is_tombstoned(&repo_ref).unwrap_or(false) => {}
Err(err) => return Err(internal(err)),
}
if options.purge {
let (before, after) = self.stores.purge_repo(&repo_ref).map_err(|err| {
ControlError::Internal(format!("removing the repository's directory: {err:#}"))
})?;
return Ok(gunnar_grpc::seam::DeleteOutcome::Purged {
bytes_reclaimed: before.saturating_sub(after),
});
}
let retention = std::time::Duration::from_secs(
u64::from(
options
.retention_days
.unwrap_or(crate::stores::DEFAULT_DELETE_RETENTION_DAYS),
) * 86_400,
);
let record = self
.stores
.soft_delete(&repo_ref, was.as_ref(), &options.principal_id, retention)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
let bytes_kept = self
.stores
.repo_dir(&repo_ref)
.map(|dir| crate::stores::dir_bytes(&dir))
.unwrap_or(0);
Ok(gunnar_grpc::seam::DeleteOutcome::Tombstoned {
purge_after_unix_ms: record.purge_after_unix_ms.unwrap_or_default(),
bytes_kept,
})
}
/// **Undelete: the row back first, the tombstone off second.**
///
/// That order is not tidiness. A crash between the two leaves a repository
/// that is catalogued *and* still fenced off, which an operator can fix
/// with a second `undelete`; the other order leaves a name with no
/// tombstone and no row, which the next `repo create` would adopt — and
/// adopting a store a create did not make is exactly the 2026-08-30
/// resurrection.
fn undelete_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_some()
{
return Err(ControlError::AlreadyExists(format!(
"{path} is in the catalog already; it is not deleted"
)));
}
let spec = self
.stores
.undelete_spec(&repo_ref)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
// `adopt`: the store on disk is the one this repository always had.
// This is the second of the two cases where adopting is right rather
// than a resurrection (the first is `archive-in`), and it is right
// because the tombstone says so and carries the hash kind the adopt
// path checks the store's marker against.
let repo = self
.stores
.create_repo(spec, true)
.map_err(|err| ControlError::Internal(format!("restoring {path}: {err:#}")))?;
self.stores
.clear_tombstone(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
Ok(repo_out(&repo))
}
fn purge_repo(&self, path: &RepoPath) -> ControlResult<u64> {
let repo_ref = Self::repo_ref(path)?;
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_some()
{
return Err(ControlError::FailedPrecondition(format!(
"{path} is a LIVE repository, not a deleted one. `repo delete {path} --purge` \
deletes and reclaims it in one step; purge alone only ends a retention window."
)));
}
let tombstone = self
.stores
.tombstone_of(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
if tombstone.is_none() {
return Err(ControlError::NotFound(format!(
"{path} is neither catalogued nor tombstoned; there is nothing to purge"
)));
}
let (before, after) = self
.stores
.purge_repo(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
Ok(before.saturating_sub(after))
}
/// **`Repos.Gc` — the console's door onto the maintenance the CLI has
/// always had.**
///
/// The whole body is a resolve and a call: the maintenance itself is
/// [`crate::admin::gc_one`], the same function `gunnar gc` runs, so there
/// is one writer and not a second implementation to drift (one implementation).
///
/// Unlike [`Self::purge_repo`] this one demands the repository be **live**.
/// A tombstoned store is on its way out and sealing it would spend the I/O
/// of a compaction on bytes a sweep is about to unlink; that is `Purge`'s
/// job, and saying so by name is better than maintaining a corpse.
fn gc_repo(&self, path: &RepoPath) -> ControlResult<gunnar_grpc::seam::GcJob> {
let repo_ref = Self::repo_ref(path)?;
// **The existence check is here and the resolve is in the runner**, and
// that is not a duplicate. This one refuses a name that was never a
// repository *now*, so a typo comes back as NOT_FOUND on the request
// that made it rather than as a FAILED job somebody has to go and read;
// the runner's is what happens when a repository that WAS live is gone
// by the time its turn comes, minutes later.
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!(
"{path} is not a live repository; nothing to gc"
)));
}
Ok(gc_job_out(path, self.gc.enqueue(&repo_ref.to_string())))
}
fn gc_status(&self, path: &RepoPath) -> ControlResult<Option<gunnar_grpc::seam::GcJob>> {
let repo_ref = Self::repo_ref(path)?;
Ok(self
.gc
.status(&repo_ref.to_string())
.map(|job| gc_job_out(path, job)))
}
// ── accounts: the gate, and the per-user hard purge ────────────────────
fn gate_account(
&self,
account: &str,
actor: &str,
note: &str,
) -> ControlResult<(gunnar_grpc::seam::AccountGate, bool)> {
// Read before write, so `was_already_gated` reports what was there
// rather than what this call put there. Not atomic with the write, and
// it does not need to be: the field is a courtesy to a console, and the
// worst a lost race costs is a console saying "you locked this" about a
// lockout that arrived a millisecond earlier. The *lockout* is atomic,
// because `AuthTable::gate` is one compare-and-swap.
let before = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.is_some();
let gate = self
.stores
.auth()
.gate(gunnar_core::AccountGate::administrative(
account, actor, note,
))
.map_err(gate_error)?;
tracing::warn!(
target: "gunnar::audit",
event = "account.gated",
account = %account,
actor = %actor,
note = %note,
already = before,
"an account was locked out: every level, both doors, until it is ungated"
);
Ok((gate_out(gate), before))
}
fn ungate_account(
&self,
account: &str,
) -> ControlResult<Option<gunnar_grpc::seam::AccountGate>> {
let lifted = self
.stores
.auth()
.lift_gate(account)
.map_err(internal)?
.map(gate_out);
if let Some(gate) = &lifted {
tracing::warn!(
target: "gunnar::audit",
event = "account.ungated",
account = %account,
held_by = %gate.holder,
"an account lockout was lifted"
);
}
Ok(lifted)
}
fn account_gates(&self) -> ControlResult<Vec<gunnar_grpc::seam::AccountGate>> {
Ok(self
.stores
.auth()
.gates()
.map_err(internal)?
.into_iter()
.map(gate_out)
.collect())
}
/// **Gate first, enqueue second — and read the gate BACK.**
///
/// The order is the whole guarantee: by the time this returns the account
/// is shut, so there is no window in which a purge has been accepted and
/// the user can still push into the account it is about to walk.
///
/// The gate is then read back out of the table rather than the one just
/// written being returned, and that is not belt-and-braces — it is the only
/// correct answer in two cases this call cannot distinguish up front:
///
/// * the account was **already** under an administrative gate. The lease is
/// not written over it (`gate_for_purge` refuses to downgrade), so the
/// gate in force is somebody else's and the job will not lift it.
/// * the account already had a purge **queued or running**. The queue
/// returns that job instead of this one, so the lease that will be
/// released is the *other* job's, and returning this call's freshly
/// minted id would name a lockout nothing is going to lift.
fn purge_account(
&self,
account: &str,
options: gunnar_grpc::seam::PurgeAccountOptions,
) -> ControlResult<(gunnar_grpc::seam::PurgeJob, gunnar_grpc::seam::AccountGate)> {
let account = account.trim();
if account.is_empty() {
return Err(ControlError::InvalidArgument(
"an account purge needs an account name".into(),
));
}
// **A machine's account is never purged** (owner ruling g-nonpay 2026-09-29:
// "machine principals ... are never touched"). Only a person's account holds a
// customer's data; a machine, a deploy key or a group is the estate itself —
// the twin, the billing principal, a CI bot — and emptying it is never the
// answer to a question about money. Refused before the lockout: nothing is
// locked and nothing is queued.
if let Some(p) = self.stores.auth().principal(account).map_err(internal)? {
if p.kind != gunnar_core::PrincipalKind::User {
return Err(ControlError::FailedPrecondition(format!(
"{account} is a machine principal ({:?}), not a person: a machine's account is never \
purged. Remove the principal itself if it is really to go.",
p.kind
)));
}
}
let job_id = self.purge.next_job_id();
// **The lockout, before anything else.** A failure here is a refusal
// and nothing is enqueued: a purge that could not lock the account is
// a purge that would race every reader in it, and running it anyway
// would be worse than not running it.
self.gate_for_purge(account, &job_id, &options)?;
let order = crate::purge_queue::PurgeOrder::new(account, &job_id, &options.reason)
.by(&options.principal_id)
.removing_principal(options.remove_principal)
.expecting_empty(options.expect_empty);
let job = self.purge.enqueue(order);
let gate = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.map(gate_out)
// The account is gated — this call just gated it — so an absence
// here means the row went between the write and the read, which is
// an operator ungating an account mid-request. Reported as the
// fact it is rather than as a gate that is not there.
.unwrap_or_else(|| gunnar_grpc::seam::AccountGate {
account: account.to_owned(),
reason: gunnar_grpc::seam::GateReason::Purge,
holder: job_id.clone(),
actor: options.principal_id.clone(),
note: "the lockout was lifted between being taken and being read back".into(),
at_unix_ms: gunnar_core::now_ms(),
expires_unix_ms: 0,
});
Ok((purge_job_out(job), gate))
}
fn purge_account_status(
&self,
account: &str,
) -> ControlResult<(
Option<gunnar_grpc::seam::PurgeJob>,
Option<gunnar_grpc::seam::AccountGate>,
)> {
let job = self.purge.status(account).map(purge_job_out);
let gate = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.map(gate_out);
Ok((job, gate))
}
fn default_delete_retention_days(&self) -> u32 {
crate::stores::DEFAULT_DELETE_RETENTION_DAYS
}
fn list_deleted(
&self,
namespace: Option<&str>,
) -> ControlResult<Vec<gunnar_grpc::seam::DeletedRepoRow>> {
Ok(self
.stores
.tombstones()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.into_iter()
.filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
.map(|row| gunnar_grpc::seam::DeletedRepoRow {
path: RepoPath {
namespace: row.path.namespace().to_owned(),
repo: row.path.name().to_owned(),
},
deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
principal_id: row.tombstone.principal_id,
bytes: row.bytes,
})
.collect())
}
/// **The store directories with no catalog row and no tombstone.**
///
/// It walks the disk rather than asking the catalog — the same reason
/// [`Self::list_deleted`] does, one step further: a tombstone at least
/// exists as a file, and an orphan is defined by the absence of *every*
/// record. On the appliance 2026-08-30 that absence made ~25-39 GB at
/// `alice/linux` unreachable by `delete`, `purge`, `undelete`, the sweep
/// AND `list-deleted`, while `create` refused the name because the bytes
/// were there.
fn list_orphans(
&self,
namespace: Option<&str>,
) -> ControlResult<Vec<gunnar_grpc::seam::OrphanRow>> {
Ok(self
.stores
.orphans()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.into_iter()
.filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
.map(|row| gunnar_grpc::seam::OrphanRow {
path: RepoPath {
namespace: row.path.namespace().to_owned(),
repo: row.path.name().to_owned(),
},
dir: row.dir.display().to_string(),
bytes: row.bytes,
hash_marker: row.hash_marker.unwrap_or_default(),
modified_unix_ms: row.modified_unix_ms,
failed_reclaim: row.failed_reclaim.unwrap_or_default(),
})
.collect())
}
/// **Reclaim an orphan.** `FailedPrecondition` when it is not one, because
/// "this is a live repository" is a fact the caller can act on and not an
/// internal fault.
fn reclaim_orphan(&self, path: &RepoPath) -> ControlResult<u64> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.reclaim_orphan(&repo_ref)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn index_progress(&self) -> ControlResult<Vec<gunnar_grpc::seam::IndexProgressRow>> {
Ok(self
.stores
.engine()
.index_progress()
.into_iter()
.filter_map(|(dir, s)| {
let (namespace, name) = self.stores.repo_of_dir(&dir)?;
Some(gunnar_grpc::seam::IndexProgressRow {
namespace,
name,
pack_id: s.pack_id,
objects_done: s.objects_done,
objects_total: s.objects_total,
bytes_done: s.bytes_done,
bytes_total: s.bytes_total,
phase: if s.active {
s.phase.to_string()
} else {
"idle".to_string()
},
started_at_unix_ms: s.started_at_ms as i64,
resumed: s.resumed,
reinflations: s.reinflations,
pauses: s.pauses,
})
})
.filter(|r| r.phase != "idle")
.collect())
}
fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
let repo = self
.stores
.catalog()
.set_default_branch(&repo_ref, default_branch)
.map_err(internal)?;
// The catalog is the authority; the repository's `HEAD` is a projection
// of it and does not move on its own. Without this line the call
// answers with the new branch and `repo get` reads it back, while the
// reference advertisement — which reads `HEAD`, not the catalog — keeps
// naming the old one. That was the whole of the 2026-08-06 defect: a
// `master` donor pushed cleanly into a repository nobody could clone,
// and `set-default-branch` looked like it had fixed it.
self.stores.repoint_head(&repo).map_err(|err| {
ControlError::Internal(format!("pointing HEAD at the new default branch: {err:#}"))
})?;
Ok(repo_out(&repo))
}
fn set_visibility(
&self,
path: &RepoPath,
visibility: gunnar_proto::Visibility,
) -> ControlResult<Repo> {
if visibility == gunnar_proto::Visibility::Unspecified {
return Err(ControlError::InvalidArgument(
"visibility unspecified".into(),
));
}
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.set_visibility(&repo_ref, visibility_in(visibility)?)
.map(|repo| repo_out(&repo))
.map_err(internal)
}
/// **A member's ONE encryption key** (CRYPT-YUBI 2026-10-26): a repository's ring is
/// wrapped to the person's encryption public key, never to an SSH key. Empty when they
/// have none yet.
fn recipient_keys(&self, principal_id: &str) -> ControlResult<Vec<gunnar_proto::RecipientKey>> {
let public = self
.stores
.encryption_keys()
.public_of(principal_id)
.map_err(key_refusal)?;
Ok(public
.into_iter()
.map(|public| gunnar_proto::RecipientKey {
fingerprint: gunnar_core::auth::fingerprint_of_blob(public.as_bytes()).to_string(),
algo: if public.starts_with("age1pq1") {
gunnar_core::auth::PQ_RECIPIENT_ALGO.to_owned()
} else {
"X25519".to_owned()
},
openssh: public,
})
.collect())
}
fn encryption_key(
&self,
principal_id: &str,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
Ok(encryption_key_out(
self.stores
.encryption_keys()
.get(principal_id)
.map_err(key_refusal)?,
))
}
fn create_encryption_key(
&self,
principal_id: &str,
public_key: &str,
first: gunnar_proto::UnlockMethod,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let first = method_in(first)?;
let key = self
.stores
.encryption_keys()
.create(
principal_id,
public_key,
first,
gunnar_core::now_ms().max(0) as u64,
)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn add_unlock_method(
&self,
principal_id: &str,
method: gunnar_proto::UnlockMethod,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let method = method_in(method)?;
let key = self
.stores
.encryption_keys()
.add_method(principal_id, method, gunnar_core::now_ms().max(0) as u64)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn remove_unlock_method(
&self,
principal_id: &str,
id: &str,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let key = self
.stores
.encryption_keys()
.remove_method(principal_id, id)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.set_encrypted(&repo_ref, encrypted)
.map(|repo| repo_out(&repo))
.map_err(internal)
}
/// **`Repos.ArchiveOut`: the same bytes `gunnar repo archive-out` writes,
/// into a sink the control plane owns** — so a cron on another machine can
/// pull a backup off a box that has no shell.
///
/// It does **not** go through [`Self::open`]. An archive is three files
/// read out of a directory; taking the handle cache and the store's lock
/// for it would make a backup contend with serving, and the whole reason
/// there is no `freeze` verb is that it does not have to. The pack is
/// append-only and fsynced below the cut, so what is read is durable and
/// what arrives after the cut belongs to the next one.
#[cfg(feature = "engine-znippy")]
fn archive_out(
&self,
path: &RepoPath,
from_generation: u64,
sink: &mut dyn std::io::Write,
) -> ControlResult<gunnar_grpc::seam::ArchiveCut> {
let repo_ref = Self::repo_ref(path)?;
let dir = self
.stores
.repo_dir(&repo_ref)
.map_err(|err| ControlError::Internal(format!("resolving {path}: {err:#}")))?;
if !dir.is_dir() {
return Err(ControlError::NotFound(format!(
"{path} has nothing on disk at {}",
dir.display()
)));
}
let receipt =
crate::storage::znippy_engine::archive_out_to(&dir, &mut { sink }, from_generation)
.map_err(archive_error)?;
// **The second bytes-out counting point, and the only one outside the
// git transport.** `Repos.ArchiveOut` streams a repository's archive
// over gRPC and never touches `transport::serve_blocking`, so its bytes
// are in neither `ServerMetrics::bytes_out` nor — before this line —
// any tenant's meter. A replica pulling a hundred gigabytes would have
// read as a silent namespace.
//
// It counts as a FETCH: a caller asked for objects and objects were
// served, which is what `fetches` means. The principal is not in scope
// at this seam — the gRPC caller's identity stops at the service layer —
// so this transfer moves bytes in nobody's name, which is a smaller
// untruth than attributing it to whoever pushed last.
self.stores.flow().record(
&path.namespace,
gunnar_core::now_ms(),
&gunnar_core::Transfer {
bytes_out: receipt.bytes_out,
fetched: true,
..gunnar_core::Transfer::default()
},
);
Ok(gunnar_grpc::seam::ArchiveCut {
from_generation: receipt.from_generation,
to_generation: receipt.to_generation,
bytes_out: receipt.bytes_out,
file_sha256: receipt.file_sha256,
})
}
// ── refs and objects ───────────────────────────────────────────────────
fn list_refs(
&self,
path: &RepoPath,
prefix: &str,
peel_them: bool,
) -> ControlResult<Vec<RefTarget>> {
let (_, store) = self.open(path)?;
let git = store.serve();
// **One `refs()` call is one snapshot.** The prefix is filtered here
// rather than pushed into the store because `GitOps::refs` takes no
// filter — it hands back the ref namespace, and narrowing it is the
// caller's. See `gunnar_wire::refs`.
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let mut out = Vec::new();
for row in snapshot.with_prefix(Some(prefix)) {
out.push(ref_target_out(
git,
store.hash_kind(),
&snapshot,
row,
peel_them,
)?);
}
Ok(out)
}
fn get_ref(
&self,
path: &RepoPath,
ref_name: &str,
peel_them: bool,
) -> ControlResult<RefTarget> {
let (_, store) = self.open(path)?;
let git = store.serve();
// `FullName::new` was the validator and it died with `RefStore`; the
// rules live in `gunnar_store::refname` now, one authority for the
// whole tree. It is still checked here rather than trusted: this is a
// caller's string, and `refname::validate` is a path-traversal boundary
// wherever a name can become a lookup.
gunnar_store::refname::validate(ref_name)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
// `refname::validate` admits exactly `HEAD` and `refs/…`, and `HEAD` is
// **not** one of the rows (the `GitServe` HEAD ruling) — so asking for it in the snapshot
// would answer `NotFound` for a reference every repository has. The
// accessor pair is where it comes from; its target still resolves
// against the snapshot, so both branches answer out of one snapshot.
let head_row;
let row = if ref_name == gunnar_store::HEAD {
head_row = git
.head()
.map_err(store_error)?
.ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?;
&head_row
} else {
snapshot
.get(ref_name)
.ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?
};
ref_target_out(git, store.hash_kind(), &snapshot, row, peel_them)
}
fn log(
&self,
path: &RepoPath,
start: &str,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
self.log_walk(path, LogStart::One(start), limit, touching)
}
/// Every branch and tag tip, and `HEAD`, walked as one history: the
/// History page's "all branches" view (owner order 2026-10-18), so a
/// commit that lives only on a side branch is drawn too.
fn log_all_refs(
&self,
path: &RepoPath,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
self.log_walk(path, LogStart::AllRefs, limit, touching)
}
fn create_pull(
&self,
path: &RepoPath,
author: &str,
title: &str,
description: &str,
source_ref: &str,
target_ref: &str,
) -> ControlResult<PullRequest> {
for (what, name) in [("source_ref", source_ref), ("target_ref", target_ref)] {
if !name.starts_with("refs/heads/") {
return Err(ControlError::InvalidArgument(format!(
"{what} must be a full branch name under refs/heads/, got {name:?}: a pull \
request merges branches, and a short name stored today is an ambiguity \
resolved differently tomorrow"
)));
}
}
if source_ref == target_ref {
return Err(ControlError::InvalidArgument(
"a branch cannot be merged into itself".to_owned(),
));
}
if title.trim().is_empty() {
return Err(ControlError::InvalidArgument(
"a title is required".to_owned(),
));
}
let repo_ref = Self::repo_ref(path)?;
let (_, store) = self.open(path)?;
let git = store.serve();
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let source = snapshot
.resolve(source_ref)
.map_err(wire_error)?
.ok_or_else(|| ControlError::NotFound(format!("{source_ref} in {path}")))?;
snapshot
.resolve(target_ref)
.map_err(wire_error)?
.ok_or_else(|| ControlError::NotFound(format!("{target_ref} in {path}")))?;
let record = self
.pulls()?
.create(
&repo_ref.to_string(),
author,
title,
description,
source_ref,
target_ref,
&source.to_string(),
)
.map_err(internal)?;
Ok(pull_out(path, record))
}
fn list_pulls(&self, path: &RepoPath, state: i32) -> ControlResult<Vec<PullRequest>> {
let repo_ref = Self::repo_ref(path)?;
// Existence check: a list for a repository that is not there is
// NotFound, never an empty success.
self.get_repo(path)?;
let filter = match PullRequestState::try_from(state) {
Ok(PullRequestState::Unspecified) | Err(_) => None,
Ok(PullRequestState::Open) => Some(gunnar_core::PullState::Open),
Ok(PullRequestState::Merged) => Some(gunnar_core::PullState::Merged),
Ok(PullRequestState::Closed) => Some(gunnar_core::PullState::Closed),
};
Ok(self
.pulls()?
.list(&repo_ref.to_string(), filter)
.map_err(internal)?
.into_iter()
.map(|record| pull_out(path, record))
.collect())
}
fn get_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
let repo_ref = Self::repo_ref(path)?;
self.get_repo(path)?;
self.pulls()?
.get(&repo_ref.to_string(), number)
.map_err(internal)?
.map(|record| pull_out(path, record))
.ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))
}
fn close_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
let repo_ref = Self::repo_ref(path)?;
self.get_repo(path)?;
match self
.pulls()?
.close(&repo_ref.to_string(), number)
.map_err(internal)?
{
gunnar_core::PullTransition::Done(record) => Ok(pull_out(path, record)),
gunnar_core::PullTransition::NotFound => Err(ControlError::NotFound(format!(
"pull request #{number} in {path}"
))),
gunnar_core::PullTransition::NotOpen(state) => Err(ControlError::FailedPrecondition(
format!("pull request #{number} is already {state:?}, not open"),
)),
}
}
/// A fast-forward, a merge commit or a squash (GUN-MERGE-ENGINE, 2026-11-01), and
/// every step is ordered so a failure leaves the truth on disk: the ancestry is
/// decided by the same [`gunnar_policy::ancestry::is_descendant_of`] branch
/// protection trusts (fail-closed: inconclusive is a refusal, never an allow), a
/// merge commit is made by [`crate::merge_engine`] and judged by every gate before
/// it is stored, the ref moves through the store's own compare-and-swap — the
/// primitive receive-pack uses, so this is not a second ref-write path — and the
/// record is marked merged only AFTER the ref has moved.
fn merge_pull(
&self,
path: &RepoPath,
number: u64,
principal_id: &str,
options: &gunnar_grpc::MergeOptions,
) -> ControlResult<(PullRequest, MergedRef)> {
self.merge_or_check(path, number, principal_id, options, false)
}
/// Every check of [`Self::merge_pull`], and no write (GUN-FIELD-TEST, 2026-11-01).
fn merge_signing_key(&self) -> ControlResult<(String, String)> {
// Read first: a box that takes no write (a twin, a browse replica) never mints.
let public = match self
.stores
.merge_signing_public()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
{
Some(public) => public,
None => {
if let Some(why) = self.frozen() {
return Err(ControlError::Unavailable(format!(
"{why}: no merge-signing key here yet; ask the primary"
)));
}
The file goes on; the console shows all of it.