gunnar-server 3.0.0

//! Where the server's state lives on this box, and the one place a repository
//! is turned into a pair of stores.
//!
//! ```text
//!   <data-root>/
//!     control.redb            the catalog, the identity table, the issuer
//!                             certificate AND the sealed secrets (the certificate-identity plan)
//!     secrets.kek             the key-encrypting key — a file, not a row
//!     control.git/            LEGACY. The Arrow secret log, left where it is.
//!     repos/<ns>/<name>.git/  one bare repository per catalog entry
//! ```
//!
//! # One control store, and the migration that made it one
//!
//! The catalog and the identity table used to be two JSON blobs behind two
//! references *inside* `control.git`. They moved to redb because the storage
//! engine takes objects as packs and refuses a single-object write, which is
//! exactly what a control document is — `gunnar_core::control` has the argument.
//!
//! What did **not** move then was `RepoStore::secrets()`, the Arrow log the
//! sealed SSH host key lived in, and this header used to say why: *"folding it
//! in then would have been a second migration riding on the first"*. It
//! also named what deferring it cost — the control **repository** was still
//! opened, so a `gunnar-server` still needed a storage engine to read its own
//! host key, and the only mitigation was to open it **lazily** so that at least
//! the catalog and the identity table stayed reachable without one.
//!
//! **That second migration happened on 2026-08-10.** Secrets are rows in
//! `control.redb` ([`gunnar_core::secrets`]), the lazy `OnceLock` is gone, and
//! so is [`RepoStore::secrets`] and the per-repository Arrow log every
//! `open_repo` used to create beside a repository nobody read secrets from.
//! `ServerStores` no longer opens a repository for any reason except serving
//! one.
//!
//! # 🚨 The live SSH host key crosses this, and it is not a flag day
//!
//! `serve.rs` calls `ensure_host_key` on every boot. A migration that lost the
//! key would mint a fresh one and hand every existing client
//! `REMOTE HOST IDENTIFICATION HAS CHANGED`, which is what an attack looks like.
//! So [`ServerStores::secrets`] is a **read-old / write-new** store
//! ([`gunnar_core::MigratingSecretStore`]): redb first, the Arrow log as the
//! fallback, the adopted value written back into redb, and **the Arrow log left
//! on disk byte for byte**. An appliance that boots the new binary keeps its
//! fingerprint; one that is rolled back to the old binary still finds its key
//! where it left it. Nothing here deletes old data.
//!
//! The legacy log is opened only if `secrets.arrow` is already there
//! ([`gunnar_core::legacy_secret_log`]), so a fresh installation never creates
//! one, and it is opened as a **plain file through `gunnar_core::arrow_log`** —
//! no engine is involved, and never needed to be.
//!
//! # One handle per repository, and a bound on how many
//!
//! An engine's store takes a writer lock on its own directory — the znippy arm
//! caches one `GitStore` per path *under* the lock and says why (*"two
//! instances over one directory would be two writers with two tails"*) — so two
//! handles on the same git directory must not exist at once. Opening one per
//! request would therefore fail the moment two clones of the same repository
//! overlapped. Handles are cached instead, which is also what the
//! single-writer requirement of the Arrow-IPC ref log wants.
//!
//! **Correction, 2026-08-10.** This paragraph used to make that argument about
//! `FsStore`'s `O_EXCL` lock on **its Arrow secret log**. That lock is gone with
//! the log: a repository no longer has a secret store, so nothing about
//! *secrets* constrains how many handles it can have. The constraint is real and
//! it is the engine's own writer lock; only the citation was stale.
//!
//! Until 2026-08-03 that cache had no bound and no eviction, so a server that
//! touched a million repositories accumulated a million open handles: a lock
//! file, an object store with its mmapped packs, and its logs, each. File
//! descriptors run out long before memory does, and when they do *every* open
//! fails while the cache holds descriptors for repositories nobody has touched
//! in weeks.
//!
//! It is now a bounded LRU, and the two halves of that are equally load-bearing:
//!
//! * **the bound**: [`DEFAULT_OPEN_REPOS`] handles, evicting the least recently
//!   used on insert; and
//! * **liveness**: the value is an `Arc<OpenRepo>`, and an entry whose
//!   handle somebody else still holds is **skipped by the evictor**. Dropping it
//!   from the map would not close the in-flight transfer (the `Arc` keeps it
//!   alive), but it would let the *next* request for that repository try to open
//!   a second store on a git directory whose writer lock is still held, turning
//!   a capacity policy into a spurious "the store is locked" failure in the
//!   middle of a clone.
//!
//! The map is behind one mutex, which is still the server's one global
//! serialisation point. The critical section holds no I/O (opening a store
//! happens outside it), but a sharded map is the next thing to do here if
//! contention ever shows up in a profile. It has not been measured, so it has
//! not been built.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::{Arc, Mutex, MutexGuard};

use anyhow::{Context, Result};
use gunnar_core::{AuthTable, ControlStore, HashKind, NewRepo, Repo, RepoCatalog, RepoRef};
use gunnar_ssh::Control;

use crate::storage::OpenRepo;

/// **Legacy, and kept.** The bare repository that held the catalog and the
/// identity table until the certificate-identity plan moved them to [`CONTROL_DB`], and the sealed
/// SSH host key until 2026-08-10 moved that there too.
///
/// It is still named, and still created, for two reasons that have nothing to
/// do with git: [`gunnar_core::Kek::for_control_dir`] keeps `secrets.kek` in it
/// — a file beside the data, deliberately not a row inside it — and
/// [`gunnar_core::legacy_secret_log`] looks under it for the Arrow log an
/// existing appliance's host key is in. **No storage engine opens it any more.**
pub const CONTROL_DIR: &str = "control.git";

/// **The control database**: the catalog, the identity table, the issuer
/// certificate and the sealed secrets (the certificate-identity plan).
pub const CONTROL_DB: &str = "control.redb";

/// The directory every served repository lives under.
pub const REPOS_DIR: &str = "repos";
/// Where the transaction log lives under the data root.
pub const TXLOG_DIR: &str = "txlog";

/// **The identity half of the metadata mirror**, under
/// `<slow>/metadata/` — principals and enrolled public keys, and nothing
/// secret.
///
/// A mirror and never a source: nothing reads it back into the control store.
/// It is there for the morning after the stripe is gone, when the question is
/// *who had access to this box, and with which keys* — and `control.redb` is on
/// the four disks that just died together.
pub const IDENTITY_MIRROR: &str = "identity.json";

/// The hash kind the **legacy** Arrow secret log under [`CONTROL_DIR`] was
/// opened with, and therefore the one it must be reopened with to be read.
///
/// Nothing git-shaped was ever stored in that log — `ArrowLogStore` records a
/// hash kind and refuses a reopen that disagrees with it (`Error::
/// HashKindMismatch`), so this is a compatibility constant and not a choice
/// being made afresh. sha256 because that is gunnar's default and what
/// every control repository was opened with.
pub const CONTROL_HASH_KIND: gix_hash::Kind = gix_hash::Kind::Sha256;

/// How many repository handles are kept open at once.
///
/// Each one costs several file descriptors: the engine's writer lock and
/// whatever its object store has open — mmapped packs and a redb tail on the
/// znippy arm, loose objects and packs on the gix one. (It was "ten, including
/// the two Arrow logs" until 2026-08-10; a repository has no Arrow logs any
/// more.) So this is a descriptor budget in disguise. 256 sits an order of magnitude
/// under a default 1024-descriptor `RLIMIT_NOFILE` while still keeping every
/// repository an ordinary server touches in a minute resident.
///
/// It is a *cache* bound, not a concurrency limit: a repository whose handle is
/// in use is never evicted, so exceeding it costs a re-open, never a failure.
///
/// **It is the CEILING, not the number** (2026-08-29). "256 sits an order of
/// magnitude under a default 1024-descriptor `RLIMIT_NOFILE`" was arithmetic
/// on handles, and a handle is five descriptors on the znippy engine — 256
/// handles are 1280 descriptors, over the 1024 the appliance's PID 1 ran with.
/// The rig hit `EMFILE` on the 202nd repository. [`ServerStores::open`] now
/// takes [`crate::limits::open_repo_budget`] of the limit the process actually
/// has: this value under a comfortable limit, 153 under 1024.
pub const DEFAULT_OPEN_REPOS: usize = 256;

/// Below this share of `MemTotal` still available, the handle cache gives
/// handles back ([`ServerStores::reclaim_under_pressure`]).
///
/// Twenty-five percent, one step **above** the indexer watchdog's 20 % floor
/// (`ZNIPPY_GIT_MEM_BUDGET_PCT`, the memory limit) and
/// deliberately so: the cache is the cheaper thing to give up, and it should
/// have given some back before the watchdog starts pausing pushes. The 2 GB box's
/// 2 GB box reached the watchdog's floor with 256 handles still held and no way
/// to reach them.
pub const RECLAIM_FLOOR_PERCENT: u64 = 25;

/// How often the memory behind the handle cache is probed, at most.
///
/// A quarter second: fast enough that a box falling into pressure gives handles
/// back within a few opens, slow enough that a `/proc/meminfo` read is never on
/// a hot path. It is taken only on a cache miss, which already costs a store
/// open.
pub const RECLAIM_PROBE_INTERVAL: std::time::Duration = std::time::Duration::from_millis(250);

/// **How long a soft-deleted repository's bytes are kept before a sweep
/// reclaims them.** Thirty days, which is the ordinary expectation of a paid
/// forge and the number the drill states.
///
/// It is a DEFAULT and not a rule: every delete carries its own
/// `purge_after_unix_ms`, written into the tombstone at the moment of the
/// delete, so changing this constant never moves the deadline of a repository
/// that is already deleted. A promise made to a customer on the day they
/// deleted something must not be shortened by a later deploy.
pub const DEFAULT_DELETE_RETENTION_DAYS: u32 = 30;

/// How often [`ServerStores::sweep_expired_tombstones`] runs on a serving box.
///
/// Hourly. The window is measured in days, so the sweep's own period is noise
/// against it, and an hour is short enough that an operator who set a
/// `--retention-days 0` to reclaim space sees it happen while they are still
/// watching.
pub const TOMBSTONE_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3600);

/// **How often the flow ledger's live counters are copied into the control
/// store: every 60 seconds.**
///
/// The window this bounds is *how much of today is lost to a hard power cut*,
/// and a minute of one namespace's byte counts is the right size of loss for
/// the write it costs — one redb commit per minute for the whole box, however
/// many namespaces moved bytes, because the flush batches every row into one
/// document update. A one-second flush would buy a minute of accuracy on a
/// meter nobody bills to the minute and pay for it sixty times over; an hourly
/// one would make a reboot look like a quiet hour.
pub const FLOW_FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);

/// Why a name is tombstoned. The two cases are NOT the same and must never be
/// collapsed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TombstoneReason {
    /// **A soft delete.** The bytes are being HELD, on purpose, until
    /// `purge_after_unix_ms`. `repo undelete` puts the repository back.
    Deleted,
    /// **A reclaim that FAILED** (2026-08-30). The bytes are OWED, not
    /// held: a `remove_dir_all` did not work and the name is fenced off until
    /// `repo delete` is run again and succeeds. There is nothing to undelete —
    /// the operator asked for the bytes to go.
    ReclaimFailed,
}

impl TombstoneReason {
    /// The word a message uses.
    pub fn as_str(self) -> &'static str {
        match self {
            TombstoneReason::Deleted => "soft-deleted",
            TombstoneReason::ReclaimFailed => "reclaim failed",
        }
    }
}

/// **A dated tombstone: the record beside a repository whose catalog row is
/// gone and whose bytes are not.**
///
/// Written by [`ServerStores::soft_delete`] and by a failed
/// [`ServerStores::purge_repo`], read by [`ServerStores::refuse_stale_store`]
/// (which is what makes the NAME unusable), by
/// [`ServerStores::undelete`] and by the sweep.
///
/// # Why the catalog row is copied in here
///
/// `undelete` has to restore what was there — the hash kind above all. A
/// repository recreated `--hash sha256` over a sha1 store serves sha1 refs to a
/// client that asked for sha256 and answers `fatal: protocol error`; that is
/// the 2026-08-30 resurrection this closed, and an undelete that guessed the
/// defaults would re-open it from the other side. The store's own
/// `OBJECT_FORMAT` marker is the second opinion, and
/// [`ServerStores::create_repo`]'s adopt path checks them against each other.
///
/// # It is JSON, and it is read leniently
///
/// An earlier build wrote a free-text note here. A file that does not parse as JSON is
/// therefore read as one of those: reason `ReclaimFailed`, no deadline, the
/// text kept as the note. An old appliance's tombstone keeps working, and a
/// new one is never mistaken for a soft delete it was not.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Tombstone {
    /// When the delete was asked for.
    pub deleted_at_unix_ms: i64,
    /// When a sweep may reclaim the bytes.
    ///
    /// `None` means **never on its own**: a failed reclaim is owed to an
    /// operator, not scheduled, and a sweep that purged it would be doing the
    /// thing that already failed, unattended.
    pub purge_after_unix_ms: Option<i64>,
    /// Who asked. Empty when a local CLI did it with no principal.
    #[serde(default)]
    pub principal_id: String,
    /// Held, or owed.
    pub reason: TombstoneReason,
    /// The catalog row as it was, so `undelete` restores it rather than a
    /// default. `None` on a tombstone written before this existed.
    #[serde(default)]
    pub repo: Option<Repo>,
    /// A sentence a human reads.
    #[serde(default)]
    pub note: String,
}

impl Tombstone {
    /// Read one, leniently. See the type docs on the free-text form.
    pub fn parse(text: &str) -> Tombstone {
        if let Ok(parsed) = serde_json::from_str::<Tombstone>(text) {
            return parsed;
        }
        Tombstone {
            deleted_at_unix_ms: 0,
            purge_after_unix_ms: None,
            principal_id: String::new(),
            reason: TombstoneReason::ReclaimFailed,
            repo: None,
            note: text.trim().to_owned(),
        }
    }

    /// Is the retention window over at `now`?
    ///
    /// A tombstone with no deadline is **never** expired, whatever `now` is.
    pub fn is_expired_at(&self, now_unix_ms: i64) -> bool {
        match self.purge_after_unix_ms {
            Some(at) => now_unix_ms >= at,
            None => false,
        }
    }

    /// How long is left, in whole seconds, or `None` when there is no deadline.
    pub fn seconds_left(&self, now_unix_ms: i64) -> Option<i64> {
        self.purge_after_unix_ms
            .map(|at| (at - now_unix_ms).max(0) / 1000)
    }
}

/// One soft-deleted repository, as `repo list-deleted` and `Admin.StorageInfo`
/// report it.
#[derive(Debug, Clone)]
pub struct DeletedRepo {
    /// Where it was.
    pub path: RepoRef,
    /// The record beside it.
    pub tombstone: Tombstone,
    /// Bytes its directory still holds. **This is the number a bill is made
    /// of**: a tombstone is storage the account is still using.
    pub bytes: u64,
}

/// **A store directory with no catalog row and no soft-delete tombstone: an
/// ORPHAN**, as `repo list-orphans` reports it.
///
/// This is the class of bytes nothing on the box could see before 2026-08-30.
/// `repo list` reads the catalog and an orphan is not in it; `repo
/// list-deleted`, `repo purge`, `repo undelete` and the expiry sweep all key
/// off a `Deleted` tombstone and an orphan has none. Every reclaim path was
/// keyed off a record that had gone.
///
/// The listing invariant this type exists to make true: **every `<name>.git`
/// directory under `repos/` appears in exactly one of `repo list`, `repo
/// list-deleted` and `repo list-orphans`** — so no directory on the disk is
/// invisible to every door.
#[derive(Debug, Clone)]
pub struct OrphanStore {
    /// The name the directory stands at.
    pub path: RepoRef,
    /// Where it is, for an operator with a shell — and there is none on the
    /// appliance, which is why the other fields exist.
    pub dir: PathBuf,
    /// What it is holding.
    pub bytes: u64,
    /// The `OBJECT_FORMAT` marker, when it has one. `None` is a store written
    /// before the marker existed, and it is what `create --adopt --hash` takes
    /// the caller's word about.
    pub hash_marker: Option<String>,
    /// Newest mtime under the directory, so an operator can tell Friday's
    /// wreckage from something still being written to.
    pub modified_unix_ms: i64,
    /// A tombstone that is **not** a soft delete — a reclaim that FAILED and
    /// is owed to an operator. Those are orphans too: `tombstones()` filters
    /// them out of `list-deleted` on purpose, so without this they would be in
    /// no listing at all.
    pub failed_reclaim: Option<String>,
}

/// The bounded LRU of open repository handles.
struct HandleCache {
    capacity: usize,
    /// Repository key → its handle and the tick it was last handed out at.
    entries: HashMap<String, (Arc<OpenRepo>, u64)>,
    /// Monotonic, so "least recently used" is a comparison and not a list.
    clock: u64,
}

impl HandleCache {
    fn new(capacity: usize) -> Self {
        HandleCache {
            capacity: capacity.max(1),
            entries: HashMap::new(),
            clock: 0,
        }
    }

    /// Lower (or raise) the bound and evict down to it.
    ///
    /// Returns how many handles were dropped, so the caller can say it once
    /// rather than have this file guess at a log line's wording.
    fn set_capacity(&mut self, capacity: usize) -> usize {
        let before = self.entries.len();
        self.capacity = capacity.max(1);
        self.evict();
        before.saturating_sub(self.entries.len())
    }

    fn touch(&mut self, key: &str) -> Option<Arc<OpenRepo>> {
        self.clock += 1;
        let clock = self.clock;
        let (store, used) = self.entries.get_mut(key)?;
        *used = clock;
        Some(Arc::clone(store))
    }

    /// Insert `store` under `key` unless another thread already won that race,
    /// and return whichever handle is now the cached one.
    fn insert(&mut self, key: String, store: Arc<OpenRepo>) -> Arc<OpenRepo> {
        if let Some(existing) = self.touch(&key) {
            // The loser's handle is dropped by the caller, releasing its lock.
            return existing;
        }
        self.clock += 1;
        let clock = self.clock;
        self.entries.insert(key, (Arc::clone(&store), clock));
        self.evict();
        store
    }

    /// Drop least-recently-used entries until the bound holds again, skipping
    /// any handle somebody else is still holding.
    ///
    /// A cache that cannot get under its bound because every handle is in use
    /// says so once per eviction pass rather than growing silently: over the
    /// bound is a real state, and an operator whose descriptors are about to run
    /// out needs to hear about it before `accept` starts failing.
    fn evict(&mut self) {
        while self.entries.len() > self.capacity {
            let victim = self
                .entries
                .iter()
                // The iterator borrows rather than clones, so `1` is the
                // cache's own reference and nothing else. Anything above it is
                // a caller holding the handle through an operation.
                .filter(|(_, (store, _))| Arc::strong_count(store) == 1)
                .min_by_key(|(_, (_, used))| *used)
                .map(|(key, _)| key.clone());
            match victim {
                Some(key) => {
                    self.entries.remove(&key);
                }
                None => {
                    tracing::warn!(
                        open = self.entries.len(),
                        capacity = self.capacity,
                        "every open repository handle is in use; the cache is over its bound \
                         until one of them is released"
                    );
                    return;
                }
            }
        }
    }
}

/// Every store the server needs, opened once.
pub struct ServerStores {
    /// **The identity core of this box** (IDENTITY-HOST, 2026-10-27), set once
    /// by the server when it builds it ([`crate::identity::build`]): the paths
    /// outside a gRPC context (a pulled `sign_in_again` or `add_users` order)
    /// mint their first-login links through it. `None` before the server is up.
    /// **Weak, on purpose**: the core holds the control plane, which holds
    /// these stores; a strong handle here is a cycle that keeps the redb files
    /// open after the server stops (an in-process restart met "Database already
    /// open"). The server's context owns the core.
    identity: std::sync::OnceLock<std::sync::Weak<gunnar_grpc::identity::GunnarIdentity>>,
    root: PathBuf,
    repos_root: PathBuf,
    control_dir: PathBuf,
    /// **Which storage engine this process runs.**
    ///
    /// Resolved once, at [`ServerStores::open`], from `GUNNAR_ENGINE` or the
    /// build's default. A field and not a global: the `OnceLock` that used to
    /// hold an installed engine is gone with the crate that installed into it,
    /// and "which engine" is now ordinary configuration that arrives with the
    /// data root rather than a plugin registered from a `main` somewhere else.
    ///
    /// It is resolved *before* the first repository is opened, so a build asked
    /// for an engine it does not carry fails at startup with a message naming
    /// what it does carry — never by quietly serving from the other one.
    engine: crate::storage::Engine,
    /// **The raw secret store**, opened eagerly with everything else.
    ///
    /// It used to be a `OnceLock<Arc<dyn RepoStore>>` holding the control
    /// *repository*, opened lazily because opening it needed a storage engine
    /// and the catalog and the identity table had stopped needing one. There is
    /// nothing left to defer: this is a table in the database that is already
    /// open, plus — on an appliance that predates 2026-08-10 — a read-only
    /// handle on the legacy Arrow log, which is a plain file.
    ///
    /// Raw, not sealed. [`Self::secrets`] is the only supported way to reach it
    /// and it puts the seal on; nothing on the boot path may write a private key
    /// through this handle. It is private for exactly that reason.
    secrets: Arc<dyn gunnar_store::SecretStore>,
    /// **Whether a pre-sealing secret log is on this disk at all.**
    ///
    /// The one fact that decides whether an unsealed value read back out of
    /// [`Self::secrets`] is gunnar's own history or somebody else's writing.
    /// `legacy_secret_log` answered it once, at open, out of `secrets.arrow`'s
    /// existence; [`Self::secrets`] hands that answer to the seal as a
    /// `gunnar_core::PlaintextPolicy` rather than letting the decorator guess it
    /// from bytes that cannot carry it. `false` on every installation made after
    /// the 2026-08-10 redb migration, which is every installation that was not
    /// upgraded across it.
    legacy_secrets: bool,
    /// The key-encrypting key the control store's secrets are sealed under
    /// Resolved at open, so a KEK that cannot be read fails the boot
    /// loudly rather than at whatever moment first touches a secret. It is a
    /// file beside the database, not a row in it — which is what makes a copied
    /// `control.redb` useless — so it costs no engine either.
    ///
    /// Behind a lock (S9, 2026-10-10): a twin opens under a KEK of its own and is handed its
    /// primary's as the log's first frame ([`Self::adopt_kek`]); from that frame on it opens the
    /// rows it holds — a sold front's `Sealed.Get` on the twin — without a restart.
    kek: std::sync::RwLock<gunnar_core::Kek>,
    /// The catalog, the identity table and the issuer certificate (the certificate-identity plan).
    control_store: Arc<dyn ControlStore>,
    /// **The one redb handle** under `control_store` and `secrets` — what the
    /// runtime data-set growth takes off the stripe and puts back
    /// (`gunnar_core::ControlDb::suspend` / `resume`, `crate::grow`).
    control_db: Arc<gunnar_core::ControlDb>,
    /// **The repository stores are CLOSED for a growth** (`crate::grow`):
    /// `repo_store` refuses to open anything while this is set, so a
    /// background worker meets an error and retries rather than pinning the
    /// mount that is about to go. Never set outside a growth.
    suspended: AtomicBool,
    /// **The transaction log** every durable mutation on this box is a frame
    /// in, before it lands (`gunnar_core::txlog`). The twin's source.
    txlog: Arc<gunnar_core::txlog::Log>,
    /// This box is a twin (`<root>/twin.json`, `crate::twin`). Runtime, not decided once at open:
    /// a fresh box becomes a twin through `Bootstrap.BecomeTwin` while it runs, and from that
    /// moment it writes no per-box mirror (the KVM drill found the fresh box's own mirror left
    /// on the twin's disk).
    ///
    /// **An `Arc` since 2026-09-21 (lane T4)**, so the SSH/HTTP front door can
    /// hold the SAME flag rather than a copy taken at start-up: a box conferred
    /// the twin role over the wire must stop serving git the instant the role
    /// lands, not at the next restart.
    twin: Arc<AtomicBool>,
    /// **Snapshots being streamed from a cut** (R5.1-A, 2026-10-07; [`ServerStores::snapshot_pin`]):
    /// while any is, a gc waits, and so does the removal of a repository store or an LFS object —
    /// what the cut names must still be there when the stream reaches it.
    snapshot_pins: SnapshotPins,
    /// The control store UNDER the tee — the twin's apply path.
    raw_control_store: Arc<dyn ControlStore>,
    /// The secret store UNDER the tee — the twin's apply path.
    raw_secrets: Arc<dyn gunnar_store::SecretStore>,
    auth: AuthTable,
    catalog: RepoCatalog,
    /// Per-repository policy documents (`refs/gunnar/policy`).
    ///
    /// Opened with the other two and not lazily: it is read on the **push
    /// path**, and a store opened on first push would make the first push of
    /// every restart pay for it.
    policies: gunnar_core::PolicyStore,
    /// Pull-mirror registrations (`refs/gunnar/mirrors`).
    mirrors: gunnar_core::MirrorStore,
    /// **The flow ledger** (`refs/gunnar/flow`): what MOVED, per namespace, per
    /// UTC day.
    ///
    /// An `Arc` and not a plain field because it is the one store here that is
    /// written from the TRANSFER paths as well as read from the control plane —
    /// `transport::serve_blocking` and `gunnar_ssh::Admission` each hold a
    /// clone — and because its live half is a set of atomics that must be the
    /// same set for every holder. Two ledgers over one control store would each
    /// hold half the day and each overwrite the other's half on flush.
    ///
    /// Opened with the other documents rather than lazily, for
    /// [`ServerStores::policies`]'s reason: it is read and written on the push
    /// path, and a store opened on first push would make the first push of
    /// every restart pay for it.
    flow: Arc<gunnar_core::FlowLedger>,
    /// **When each principal last signed in and last changed something**
    /// (2026-10-21): in memory, minute-grained, flushed by its own thread in a
    /// batch — one redb transaction and one `Activity` frame per flush that
    /// moved something — never on the request path, never in the identity
    /// document. See `gunnar_core::activity`.
    activity: Arc<gunnar_core::PrincipalActivity>,
    /// **The usage ledger** (2026-10-06): every namespace's stock meters, kept
    /// current by re-measuring only what changed — `UsageBatch.ReadMany`'s
    /// source, so a billing poll is not one disk walk per tenant. Marked from
    /// this type's own mutation points (a store opening, a tombstone, a
    /// growth's resume) and from the flow ledger's observer.
    usage: Arc<crate::usage_ledger::UsageLedger>,
    /// **The server's ONE set of counters** (`gunnar_core::observe::ServerMetrics`), minted
    /// with the stores so the tasks that start before `Server::start` (the flow flush and its
    /// usage-ledger reconcile pass, A28 M17) write to the same counters the listeners do and
    /// `/metrics` publishes. `Server::start` takes this one; it mints no second set.
    metrics: Arc<gunnar_core::observe::ServerMetrics>,
    /// **Every account's audit log** (`refs/gunnar/account-audit/<account>`,
    /// 2026-10-05): the account admin's, never the operator's.
    account_audit: Arc<gunnar_core::account_audit::AccountAuditLog>,
    /// **The name filter's lists and the accounts' review queues**
    /// (`refs/gunnar/name-filter`, 2026-10-05).
    names: gunnar_core::name_filter::NameFilterStore,
    /// **The generic server settings** (`refs/gunnar/settings`, owner 2026-10-26): bytes under
    /// names the clients choose; the server knows none of them.
    settings: gunnar_core::settings::SettingsStore,
    /// **People's encryption keys** (`refs/gunnar/encryption-keys`, CRYPT-YUBI 2026-10-26): the
    /// public half of each person's one encryption key and its unlock methods' wraps, which are
    /// ciphertext to this server.
    encryption_keys: gunnar_core::encryption_key::EncryptionKeys,
    open: Mutex<HandleCache>,
    /// When the memory probe behind [`ServerStores::reclaim_under_pressure`]
    /// last ran. `None` until the first cache miss.
    last_reclaim_probe: Mutex<Option<std::time::Instant>>,
    /// **Is this box refusing client writes because it is a STANDBY?**
    ///
    /// One `Arc<AtomicBool>`, shared by clone into every [`Control`] this
    /// `ssh_control` hands out AND read/written by the replication slot's
    /// role changes — one value, so the SSH front door and the control plane
    /// can never disagree about whether a `git push` should be refused (one implementation).
    /// `true` = STANDBY (freeze non-`Machine` writes), `false` = PRIMARY.
    ///
    /// Initialised from the persisted role in `replication.json` at open, so a
    /// box that abdicated and then rebooted comes back still frozen — the same
    /// durability `demote`'s roster already has. Always present, even in a build
    /// without the `replication` feature, where it simply never turns `true`.
    writes_frozen: Arc<AtomicBool>,
    /// **Is this box a browse replica**? Sibling of
    /// `writes_frozen`, same lifecycle: set from `replication.json` at open,
    /// flipped by the replication slot on a role change, READ by the control
    /// plane's mutator refusal and by the front door's wording. A browse box is
    /// also frozen; this flag is what distinguishes it from a promotable
    /// standby.
    browse: Arc<AtomicBool>,
    /// **How many repository stores this process has opened**, cumulative and
    /// monotonic.
    ///
    /// One counter, bumped at the single site an engine store is actually
    /// opened (a [`HandleCache`] miss in [`Self::repo_store`]) — not on a cache
    /// hit, which pays nothing. Each open pays the refold cost, so a
    /// path that opens every repository on a cold box is a cost measured in
    /// this number, and it is the number a status call must not run up: see
    /// Exposed for exactly that assertion — a test proves a plain
    /// `replication list` adds ZERO to it, and `--refs` adds one per repository.
    store_opens: Arc<AtomicU64>,
    /// **Where each kind of data goes** (`crate::slow`), resolved once at open
    /// out of the layout config's `dir.<name>.path`.
    ///
    /// A field and not a global for the reason the engine is a field: which
    /// directories this server writes to is configuration that arrives with the
    /// data root, and a process-global would make two `ServerStores` in one
    /// process (the bench opens the same data root twice) share one answer.
    ///
    /// It ALWAYS resolves — to the configured paths, or to the documented
    /// default under the data root — so nothing downstream branches on whether
    /// the machine has a second spindle. What the branch is on is the CLAIM:
    /// see `crate::slow::Claim`.
    slow: crate::slow::SlowRoot,
    /// **What the installer decided**, read once at open and never re-read.
    ///
    /// Static and historical by nature: it is the only place the *reasoning*
    /// survives — which disk booted and why, which devices became `/data`,
    /// what was rejected and on what grounds. Re-reading it would not make it
    /// fresher, and an absent one stays absent with the sentence that says so.
    install: crate::layout_config::Load,
    /// **The front doors' plan-gate chain, where a mirror's worker can read
    /// it** (2026-10-02). `serve::front_control` builds the chain a push is held
    /// to — disk headroom, then the entitlement on a metered build — and
    /// installs it here, so a mirror sync is held to the SAME namespace room a
    /// push is, by the same gates, rather than being the one write on the box
    /// no ceiling holds. Empty until something installs it (a store opened by
    /// a test or a tool), which is "no ceiling", exactly the answer a chain
    /// with no gates gives.
    plan_gates: std::sync::RwLock<Vec<Arc<dyn gunnar_ssh::PlanGate>>>,
    /// **The data set's space ledger** (`gunnar_ssh::space`), installed with
    /// the plan gates by `serve::front_control` — the disk gate's own. Every
    /// door that writes a pack or an object asks it before each chunk. `None`
    /// (a test, a tool): no admission beyond the gates.
    space: std::sync::RwLock<Option<Arc<gunnar_ssh::SpaceLedger>>>,
    /// **The keys this primary was FED to enrol and has not enrolled yet**
    /// (fingerprint → OpenSSH line; `crate::fed_principals::note_fed`), read
    /// by the gRPC gate through `Identity::awaiting_enrolment`: the holder of
    /// one is told `Unavailable [key-enrolling]`, ask again, not that it is
    /// unknown (cov-E r3 2026-10-06). In memory only: the feed on disk is the
    /// authority, and every feed and every install pass recomputes it.
    fed_keys: std::sync::RwLock<std::collections::BTreeMap<String, String>>,
}

/// The longest a gc or a removal waits for a snapshot streaming from its cut
/// ([`ServerStores::wait_unpinned`]).
pub const SNAPSHOT_PIN_WAIT: std::time::Duration = std::time::Duration::from_secs(600);

/// The pin count and its condvar ([`ServerStores::snapshot_pin`]).
#[derive(Default)]
struct SnapshotPins {
    count: Mutex<usize>,
    cv: std::sync::Condvar,
}

/// Held while a snapshot streams from its cut. See [`ServerStores::snapshot_pin`].
#[must_use = "the pin holds only while the guard lives"]
pub struct SnapshotPin<'a> {
    pins: &'a SnapshotPins,
}

impl Drop for SnapshotPin<'_> {
    fn drop(&mut self) {
        let mut n = self.pins.count.lock().unwrap_or_else(|p| p.into_inner());
        *n = n.saturating_sub(1);
        self.pins.cv.notify_all();
    }
}

impl std::fmt::Debug for ServerStores {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServerStores")
            .field("root", &self.root)
            .finish_non_exhaustive()
    }
}

impl ServerStores {
    /// Open everything under `root`, creating whatever is not there yet.
    ///
    /// The handle cache is sized to the descriptor limit this process has —
    /// after [`crate::limits::nofile`] has raised it as far as it may — so the
    /// cache can never by itself put the process over `RLIMIT_NOFILE`.
    pub fn open(root: impl AsRef<Path>) -> Result<Self> {
        // `crate::budget`, not `limits::open_repo_budget` alone: the descriptor
        // term is still the descriptor one and still binds first on a small `ulimit -n`,
        // but it is now the MINIMUM of that and a memory term. The 2 GB box measured
        // why: on a 2 GB appliance the descriptor term said 256 handles, 256
        // handles held the whole box, and the indexer's watchdog then spent
        // every push failing to reclaim memory that was in the OTHER 255.
        Self::open_with_capacity(root, crate::budget::budgets().open_repos)
    }

    /// [`ServerStores::open`], with an explicit handle-cache bound.
    ///
    /// Exists so the eviction path can be tested at a capacity a test can
    /// actually exceed. A bound that is only ever exercised at 256 is a bound
    /// nothing has ever seen fire.
    pub fn open_with_capacity(root: impl AsRef<Path>, capacity: usize) -> Result<Self> {
        Self::open_with_engine(root, capacity, crate::storage::Engine::selected()?)
    }

    /// [`ServerStores::open_with_capacity`], with the engine named rather than
    /// resolved from the environment.
    ///
    /// The one seam a head-to-head bench needs: the same data root opened twice,
    /// once per engine, from one process — which is what makes an Arrow-vs-gix
    /// row a comparison rather than two anecdotes.
    pub fn open_with_engine(
        root: impl AsRef<Path>,
        capacity: usize,
        engine: crate::storage::Engine,
    ) -> Result<Self> {
        let root = root.as_ref().to_path_buf();
        let repos_root = root.join(REPOS_DIR);
        std::fs::create_dir_all(&repos_root)
            .with_context(|| format!("creating the repository root {}", repos_root.display()))?;

        let control_dir = root.join(CONTROL_DIR);
        std::fs::create_dir_all(&control_dir)
            .with_context(|| format!("creating the control directory {}", control_dir.display()))?;

        // **Where the storage is, read once — and gunnar is TOLD, never
        // discovers.** `/boot/korp/layout.conf` is what the installer wrote and
        // sealed while it still knew WHY it chose what it chose; `crate::slow`
        // obeys its `dir.<name>.path` verbatim and nothing here probes a mount
        // table to second-guess it. An appliance installed before the config
        // existed has none, which is a first-class answer carrying the sentence
        // that says so — never a refusal to start, and never an inference.
        let install = crate::layout_config::load();
        let slow = crate::slow::SlowRoot::resolve(&root, install.config());
        slow.announce(
            &crate::layout_config::config_path().display().to_string(),
            install
                .config()
                .map(|c| c.authenticity().word())
                .unwrap_or_default(),
        );
        if let Some(why) = install.absent_reason() {
            tracing::info!(
                target: crate::slow::TARGET_SLOW,
                event = "layout.absent",
                "{why}"
            );
        }
        // The panic hook in `main` has no data root; this is where it learns
        // where a crash record goes. Best effort — a crash directory that
        // cannot be made means no crash file, never a refusal to start.
        match slow.crash_dir() {
            Ok(dir) => crate::slow::register_crash_dir(dir),
            Err(err) => tracing::warn!(
                target: crate::slow::TARGET_SLOW,
                %err,
                "no crash directory on the slow disk; a panic will reach the console but \
                 will not be written down"
            ),
        }

        // The KEK for the control store's secrets. Read (or minted) here
        // so `secrets()` cannot fail on the key, and so a missing one is a
        // boot-time error with a path in it. It is a file, not a git object, so
        // this does not need an engine.
        let kek = gunnar_core::Kek::for_control_dir(&control_dir).with_context(|| {
            format!(
                "resolving the key-encrypting key for {}; secrets are sealed at rest",
                control_dir.display()
            )
        })?;

        // The certificate-identity plan. This is the whole control plane's storage, and it opens
        // whether or not this binary carries a storage engine.
        //
        // The documents and the secrets come out of ONE call because they share
        // one `redb::Database`: redb takes an exclusive file lock, so opening
        // `control.redb` twice in this process would fail rather than give two
        // views. One file, one lock, one handle.
        let db = root.join(CONTROL_DB);
        let (raw_control_store, redb_secrets, control_db) = gunnar_core::open_control_with_db(&db)
            .with_context(|| format!("opening the control store {}", db.display()))?;
        // **The transaction log**, before either store: a primary that cannot
        // write its log has no twin, and that is loud here rather than found
        // at the restore (`.nornir/twin-transaction-log.md`).
        let txlog_dir = root.join(TXLOG_DIR);
        let txlog = Arc::new(
            gunnar_core::txlog::Log::open(&txlog_dir)
                .with_context(|| format!("opening the transaction log {}", txlog_dir.display()))?,
        );
        // **Twin or primary** (`crate::twin`). A twin holds the primary's
        // numbers and originates none: its log is follower-only, nothing under
        // its root is framed, and its control plane refuses mutators (the
        // browse flag). A box mid-restore is follower-only too. A primary
        // frames its KEK FIRST — seq 1 — so a restore can open every sealed
        // secret that follows, and registers its root with the sink so the
        // signed logs and documents under it are frames before they are bytes.
        let twin = crate::twin::is_twin_root(&root);
        let restoring = crate::twin::is_restoring(&root);
        if restoring && !twin {
            // Said at every open, because the state is silent otherwise: the box serves git,
            // answers reads, and refuses every write with a message about twins.
            tracing::warn!(
                target: gunnar_core::observe::TARGET_TWIN,
                event = "twin.restoring_at_open",
                "THIS BOX IS MID-RESTORE (`txlog-restoring` is present): it originates no \
                 transactions, so every control-plane write is refused, until a restore finishes. \
                 Resend the twin's log with `gunnar control twin replay`; a finished restore that \
                 still says this lost its marker's removal to a power cut"
            );
        }
        // **An old generation of its log** (stinky row 18, the main's half): a twin
        // showed this box a newer generation signed by its own host key, so the
        // box was fenced (`crate::twin::LOG_FENCED`). A restart must not undo
        // that: it comes up follower-only, writes nothing, and says why.
        let fenced = !twin && !restoring && crate::twin::is_fenced(&root);
        if fenced {
            tracing::error!(
                target: gunnar_core::observe::TARGET_TWIN,
                event = "twin.fenced_by_generation",
                "THIS BOX IS FENCED: {}. Every write is refused and no twin is served its log.",
                crate::twin::fenced_why(&root).unwrap_or_default()
            );
        }
        if twin || restoring || fenced {
            txlog.set_follower(true);
        } else {
            // **This history's name** (2026-09-21, lane T4). A primary mints it
            // once, from the CSPRNG and from nothing on the box, and answers
            // `Mirror.LogHead` with it from then on. It is what lets a twin ask
            // "is this the same log?" instead of only "is this number bigger?"
            // — the question that catches a re-imaged main which has already
            // written PAST the twin's head, where every number still looks
            // right. A twin does not mint one: it is told its primary's, and a
            // box taking a restore is told the stream's. See `twin::LOG_ID`.
            crate::twin::mint_log_id(&root).context("minting this log's identity")?;
            if txlog.head_seq() == 0 {
                let bytes = hex::decode(kek.to_hex()).context("the KEK as bytes")?;
                txlog
                    .append(
                        gunnar_core::txlog::Kind::Kek,
                        &gunnar_core::txlog::Meta::Kek {},
                        &bytes,
                    )
                    .context("framing the KEK as the log's first transaction")?;
            }
            gunnar_core::txlog::sink::register(&root, &txlog);
            // **And the site's WireGuard hub declaration becomes a transaction**
            // (lane T11, 2026-09-21). Nothing in gunnar ever WRITES `wg/hub.conf` —
            // `korp-installer` lays it through the ESP data stage before gunnar's
            // first boot — so it has no sink call site of its own and the twin
            // never held it. A restored main with no hub declaration is a box
            // nobody can reach, which is the worst thing to find on the day the
            // twin is used.
            //
            // Framed on open, once per CHANGE and not once per boot: the digest of
            // what was last framed sits beside it in a per-box file, so a box that
            // reboots a thousand times adds one frame, not a thousand.
            //
            // **After the re-seal, not here** — see the call site below. A restored
            // box's FIRST frame after the restore must be its new KEK, and framing a
            // document before it would put the KEK second in its own log.
        }
        let control_store: Arc<dyn ControlStore> =
            Arc::new(gunnar_core::txlog::tee::LoggedControlStore::new(
                Arc::clone(&raw_control_store),
                Arc::clone(&txlog),
            ));

        // 🚨 THE LIVE SSH HOST KEY CROSSES HERE. Read-old / write-new: the
        // legacy Arrow log is consulted only for a key redb has never seen, and
        // whatever it answers is written into redb and returned unchanged. The
        // log itself is never written and never removed — an appliance rolled
        // back to a pre-migration binary still finds its host key. `None` is the
        // normal answer on any installation made after this change, and on that
        // path no Arrow log is created.
        let legacy =
            gunnar_core::legacy_secret_log(&control_dir, CONTROL_HASH_KIND).with_context(|| {
                format!(
                    "checking {} for a legacy Arrow secret log to carry the SSH host key \
                     forward from",
                    control_dir.display()
                )
            })?;
        // Recorded before the handle is type-erased: once `legacy` is inside an
        // `Arc<dyn SecretStore>` nothing downstream can ask whether there was
        // one, and `secrets()` is the caller that has to know.
        let legacy_secrets = legacy.is_some();
        let raw_secrets: Arc<dyn gunnar_store::SecretStore> =
            Arc::new(gunnar_core::MigratingSecretStore::new(redb_secrets, legacy));
        // **A restored box re-seals under a KEK of its own** before any sealed store is built
        // (`crate::twin::reseal_after_restore`). Its log is a primary's again by now.
        // Captured before the re-seal clears its marker: it names this start in what the audit
        // trail check says (`crate::twin::rematerialise_audit_trail`, below).
        let just_restored = !twin && !restoring && crate::twin::reseal_pending(&root);
        let kek = if just_restored {
            let rows = control_db
                .export()
                .context("reading the secrets to re-seal")?
                .secrets;
            crate::twin::reseal_after_restore(
                &root,
                &control_dir,
                &txlog,
                &rows,
                raw_secrets.as_ref(),
            )
            .context("re-sealing a restored box's secrets under its own KEK")?
        } else {
            kek
        };
        // **And now the site's hub declaration** (lane T11) — deliberately AFTER the re-seal.
        // A restored box's first frame after the restore is its own KEK (`reseal_after_restore`
        // frames it first so the twin following the restored box can open everything that comes
        // after), and a document framed before it would take that place. Measured: it did, and
        // `a_restored_box_reseals_every_secret_under_a_kek_of_its_own…` said so by name.
        if !twin && !restoring && !fenced {
            adopt_hub_conf(&root, &txlog);
        }
        let secrets: Arc<dyn gunnar_store::SecretStore> =
            Arc::new(gunnar_core::txlog::tee::LoggedSecretStore::new(
                Arc::clone(&raw_secrets),
                Arc::clone(&txlog),
            ));

        let auth =
            AuthTable::new(Arc::clone(&control_store)).context("opening the identity table")?;
        let catalog = RepoCatalog::new(Arc::clone(&control_store))
            .context("opening the repository catalog")?;
        let policies = gunnar_core::PolicyStore::new(Arc::clone(&control_store))
            .context("opening the policy store")?;
        let mirrors = gunnar_core::MirrorStore::new(Arc::clone(&control_store))
            .context("opening the mirror registry")?;
        let account_audit = Arc::new(
            gunnar_core::account_audit::AccountAuditLog::new(Arc::clone(&control_store))
                .context("opening the account audit logs")?,
        );
        let names = gunnar_core::name_filter::NameFilterStore::new(Arc::clone(&control_store))
            .context("opening the name filter")?;
        let settings = gunnar_core::settings::SettingsStore::new(Arc::clone(&control_store))
            .context("opening the server settings")?;
        let encryption_keys =
            gunnar_core::encryption_key::EncryptionKeys::new(Arc::clone(&control_store))
                .context("opening people's encryption keys")?;
        let flow = Arc::new(gunnar_core::FlowLedger::new(
            gunnar_core::FlowStore::new(Arc::clone(&control_store))
                .context("opening the flow ledger")?,
        ));

        tracing::info!(
            target: "gunnar::store",
            event = "store.engine_selected",
            engine = engine.name(),
            compiled = %crate::storage::Engine::compiled_list(),
            "the storage engine for this process"
        );

        // The persisted role decides whether this box comes up frozen. A
        // standby that rebooted is still a standby; a plain box (no
        // `replication.json`, or the `replication` feature off) is a writable
        // primary, which is every deployment before failover was built.
        let role_at_open = replication_role_at_open(&root);
        let frozen_at_open = role_at_open.refuses_client_writes();
        if frozen_at_open {
            tracing::warn!(
                target: "gunnar::store",
                event = "store.role_standby",
                "this box is a STANDBY (replication.json) — direct client pushes are refused; \
                 only the replication push from its primary writes. `gunnar control replication \
                 promote` makes it a writable primary."
            );
        }

        if twin {
            // Before any repository of the twin opens: the engine keeps no derived memo (C2)
            // and no exploded read cache, and resolves a pack narrow enough for the box (C1 +
            // the twin memory bound). Both are process-wide, and both used to be cmdline
            // tokens on a medium that no longer exists.
            engine.hold_no_graph_memo();
            engine.hold_twin_memory_bound();
        }
        // In the control database's own table; each flush that moved something is
        // ONE `Activity` frame on a box that originates its log (the twin's copy),
        // nothing framed on a follower. The old JSON file is read once and left.
        let activity = Arc::new(gunnar_core::PrincipalActivity::open(
            Arc::clone(&control_db),
            Some(control_dir.join(gunnar_core::ACTIVITY_FILE)),
            Some(Arc::clone(&txlog)),
        ));
        // Flushed every FLUSH_EVERY by a jera job that ends with the record (A32-14); one that
        // cannot start is said, and the record's drop still writes the last stamps.
        if let Err(e) = start_activity_flush(&activity, gunnar_core::activity::FLUSH_EVERY) {
            tracing::error!("ALERT {ACTIVITY_FLUSH} no flush ({e}): the last-seen stamps are written when the record is dropped only");
        }
        let stores = ServerStores {
            identity: std::sync::OnceLock::new(),
            root,
            repos_root,
            control_dir,
            activity,
            engine,
            secrets,
            legacy_secrets,
            kek: std::sync::RwLock::new(kek),
            control_store,
            control_db,
            suspended: AtomicBool::new(false),
            raw_control_store,
            raw_secrets,
            txlog,
            twin: Arc::new(AtomicBool::new(twin)),
            auth,
            catalog,
            policies,
            mirrors,
            flow,
            usage: Arc::new(crate::usage_ledger::UsageLedger::new()),
            metrics: Arc::new(gunnar_core::observe::ServerMetrics::new()),
            account_audit,
            names,
            settings,
            encryption_keys,
            open: Mutex::new(HandleCache::new(capacity)),
            last_reclaim_probe: Mutex::new(None),
            writes_frozen: Arc::new(AtomicBool::new(frozen_at_open || twin)),
            browse: Arc::new(AtomicBool::new(role_at_open.is_browse() || twin)),
            store_opens: Arc::new(AtomicU64::new(0)),
            slow,
            install,
            plan_gates: std::sync::RwLock::new(Vec::new()),
            space: std::sync::RwLock::new(None),
            fed_keys: std::sync::RwLock::new(std::collections::BTreeMap::new()),
            snapshot_pins: SnapshotPins::default(),
        };
        // **The flow tells the usage ledger** which namespaces moved: their
        // `*_today` meters changed, and a transfer is what grows a repository.
        {
            let usage = Arc::clone(&stores.usage);
            stores
                .flow
                .observe(gunnar_core::FlowObserver(Arc::new(move |ns: &str| {
                    usage.mark_flow(ns)
                })));
        }
        // The identity half of the metadata mirror, taken once at open so a box
        // that has not enrolled anybody since its last boot still has a current
        // copy off the stripe. The mutating verbs refresh it; see
        // `ServerStores::mirror_identity`.
        stores.mirror_identity();
        // **The audit trail, checked against the log it came from** — on EVERY open of a box
        // that originates its log, before `serve` opens it for the gate (lane T11; every start
        // since T17). The window it closes: `AuditChain::append` syncs the frame, then writes the
        // line; a primary that crashes between the two comes back with the log one entry ahead,
        // and the gate would frame a second entry at an existing seq. See
        // `crate::twin::rematerialise_audit_trail`.
        if !twin && !restoring {
            crate::twin::rematerialise_audit_trail(&stores, just_restored);
        }
        // A twin keeps no per-box mirror, including one a fresh box wrote before it became a twin.
        if stores.is_twin() {
            crate::twin::forget_mirrors(&stores);
            crate::twin::say_if_stores_were_applied(&stores);
        }
        Ok(stores)
    }

    /// **Say which fed keys this box will enrol and has not yet** — the whole
    /// set, replacing the last (`crate::fed_principals::note_fed`).
    pub fn await_enrolment(&self, keys: std::collections::BTreeMap<String, String>) {
        *self.fed_keys.write().unwrap_or_else(|p| p.into_inner()) = keys;
    }

    /// The fed key under `fingerprint` this box will enrol and has not yet, if
    /// any ([`ServerStores::await_enrolment`]).
    pub fn awaiting_enrolment(&self, fingerprint: &str) -> Option<String> {
        self.fed_keys
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .get(fingerprint)
            .cloned()
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    /// The identity core of this box, once the server built it.
    pub fn identity(&self) -> Option<std::sync::Arc<gunnar_grpc::identity::GunnarIdentity>> {
        self.identity.get().and_then(std::sync::Weak::upgrade)
    }

    /// Hand the box's identity core to the paths outside a gRPC context (once;
    /// a second hand-over is ignored: one core per box). Held weakly: the
    /// caller keeps the core alive (the server's context does).
    pub fn set_identity(&self, identity: &std::sync::Arc<gunnar_grpc::identity::GunnarIdentity>) {
        let _ = self.identity.set(std::sync::Arc::downgrade(identity));
    }

    /// **The slow disk** this server writes its off-stripe copies to.
    ///
    /// Always present. See `crate::slow` for what goes there and for the
    /// three-valued claim over it.
    pub fn slow(&self) -> &crate::slow::SlowRoot {
        &self.slow
    }

    /// **What the installer decided**, or the sentence saying there is no
    /// config. Read once at open; see the field's documentation.
    pub fn layout_config(&self) -> &crate::layout_config::Load {
        &self.install
    }

    /// **Where the control-plane audit trail lives**: `<slow>/audit/` — one of
    /// the three copies that must be on another failure domain.
    ///
    /// It used to be `<control dir>/audit-chain.jsonl`, on the stripe with the
    /// repositories whose history it records. An existing chain is ADOPTED into
    /// the new place once, on the first boot of a build that knows about the
    /// slow disk (`slow::adopt_file`: copy, fsync, unlink — `rename` answers
    /// `EXDEV` across devices, which is exactly the case that matters). If the
    /// move cannot be made the old path is returned unchanged and the server
    /// keeps writing the trail where it already is.
    ///
    /// **On a twin, or on a box mid-restore, it is a DIFFERENT file**
    /// (`audit-chain.local.jsonl`, [`gunnar_core::txlog::AUDIT_LOCAL_REL`]).
    /// Such a box holds the primary's trail as replicated state and must
    /// originate nothing into it: its own gate decisions — who browsed it, who
    /// was refused — are its own, and interleaving them into a hash chain that
    /// has exactly one writer would break every `prev` after the splice. The
    /// twin's own trail stays on the twin and is never sent anywhere.
    pub fn audit_path(&self) -> std::path::PathBuf {
        // The name is decided by the same two facts that decide whether the log originates at
        // all (`ServerStores::open`: a twin or a restoring box is a follower and registers no
        // sink), so the trail and the log can never disagree about which box this is.
        if self.is_twin() || crate::twin::is_restoring(&self.root) {
            let local = std::path::Path::new(gunnar_core::txlog::AUDIT_LOCAL_REL)
                .file_name()
                .expect("a file name")
                .to_owned();
            return match self.slow.log_dir() {
                Ok(dir) => dir.join(local),
                Err(_) => self.control_dir.join(local),
            };
        }
        let old = self.control_dir.join(gunnar_core::AUDIT_FILE);
        match self.slow.log_dir() {
            Ok(dir) => crate::slow::adopt_file(&old, &dir.join(gunnar_core::AUDIT_FILE)),
            Err(err) => {
                tracing::warn!(
                    target: crate::slow::TARGET_SLOW,
                    %err,
                    path = %old.display(),
                    "no audit directory on the slow disk; the audit trail stays on /data, where \
                     it shares a failure domain with the repositories it records"
                );
                old
            }
        }
    }

    /// **The identity half of the metadata mirror**: every principal and every
    /// enrolled public key, as one JSON document off the stripe.
    ///
    /// Public material only — a fingerprint, an `authorized_keys` line, a
    /// principal's id and name. No secret in this file, ever: the secrets are
    /// sealed in `control.redb` under a KEK that is a separate file, and a
    /// mirror that carried them would make the slow disk a second place to
    /// steal them from.
    ///
    /// Best effort, and deliberately so at every call site: this runs after a
    /// principal or a key has ALREADY been written to the control store, and
    /// failing the verb because the mirror could not be refreshed would turn a
    /// full slow disk into an outage of the identity system. A failure is one
    /// WARN naming the path.
    pub fn mirror_identity(&self) {
        // A twin keeps no mirror of its own: its identity table is a replica,
        // and a derived per-box copy on its disk is exactly the cache the twin
        // must not hold.
        if self.is_twin() {
            return;
        }
        if let Err(err) = self.try_mirror_identity() {
            tracing::warn!(
                target: crate::slow::TARGET_SLOW,
                %err,
                "the identity metadata mirror could not be refreshed; the control store itself \
                 is unaffected"
            );
        }
    }

    /// **The storage reality of this machine**, for `Admin.StorageInfo`.
    ///
    /// Three layers that are kept apart because they age differently: what the
    /// installer DECIDED (historical, from the record), what is TRUE NOW (the
    /// configured filesystems, read here), and what gunnar is CONFIGURED to do
    /// (the placement table as built). One artifact behind the first and third
    /// — `crate::layout_config` and `crate::slow` — so the console and the
    /// placement code cannot drift into two views of the same facts.
    ///
    /// **Nothing here surveys the machine.** The only paths measured are the
    /// two gunnar was told to use; no mount table is walked, no device is
    /// enumerated, and the installer's decisions are never inferred from what
    /// happens to be mounted now. A record that is missing is reported as
    /// missing.
    pub fn storage_reality(&self) -> gunnar_proto::StorageReality {
        let mut out = gunnar_proto::StorageReality {
            data_root: self.root.display().to_string(),
            slow_root: self.slow.serialdat().to_owned(),
            slow_root_source: self.slow.source().word().to_owned(),
            backing_device: self.slow.device().to_owned(),
            backing_class: self.slow.class().to_owned(),
            off_stripe: self.slow.claim().word().to_owned(),
            off_stripe_health: self.slow.health(),
            ..Default::default()
        };
        // ── 1. what the installer decided ──────────────────────────────────
        match self.install.config() {
            Some(config) => {
                out.record_present = true;
                out.record_path = config.path().display().to_string();
                out.record_version = config.version();
                // **The third state, said in the installer's own two words.**
                // `INTEGRITY ONLY` is what every appliance built today is, and
                // it is never rendered as verified.
                out.record_signature = config.authenticity().word().to_owned();
                out.record_signature_verified = config.authenticity().is_verified();
                out.record_signature_note = config.note().to_owned();
                out.record_fields = config
                    .fields()
                    .map(|(key, value)| gunnar_proto::InstallRecordField {
                        key: key.to_owned(),
                        value: value.to_owned(),
                    })
                    .collect();
            }
            None => {
                out.record_absent_reason = self
                    .install
                    .absent_reason()
                    .unwrap_or("there is no layout config")
                    .to_owned();
            }
        }
        // ── 2. what is true now ────────────────────────────────────────────
        //
        // Only the paths gunnar was TOLD to use, measured at the moment of the
        // call. No mount table is walked and no device is enumerated: a
        // filesystem gunnar was not configured with is not this call's to
        // report on.
        let mut measured: Vec<(&str, PathBuf)> = vec![("data root", self.root.clone())];
        for p in crate::slow::PLACEMENT
            .iter()
            .filter(|p| p.requires_off_stripe)
        {
            let path = self.slow.path_of(p.dir);
            if !measured.iter().any(|(_, seen)| *seen == path) {
                measured.push(("configured directory", path));
            }
        }
        out.filesystems = measured
            .iter()
            .map(|(what, path)| filesystem_use(what, path))
            .collect();
        // ── 3. what gunnar is configured to do ─────────────────────────────
        out.placement = crate::slow::PLACEMENT
            .iter()
            .map(|p| gunnar_proto::PlacementRow {
                what: p.what.to_owned(),
                path: self.slow.path_of(p.dir).display().to_string(),
                requires_off_stripe: p.requires_off_stripe,
                // The verdict is carried ONLY on the rows that need it, and it
                // is the config's per-directory answer rather than one summary
                // painted across every row. An empty string on a row that does
                // not require it is the honest shape; painting "off-stripe" on
                // git's own objects would suggest a guarantee nobody made.
                off_stripe: if p.requires_off_stripe {
                    self.slow.claim_of(p.dir).word().to_owned()
                } else {
                    String::new()
                },
                wired: p.wired,
                note: p.note.to_owned(),
            })
            .collect();
        // ── /data's own shape, and the machine at a glance (2026-10-04) ────
        //
        // `/data`'s devices come from the installer's `data.*` keys and from
        // nothing else; `backing_*` above are `/serialdat`'s and stay so.
        let config = self.install.config();
        if let Some(config) = config {
            out.data_shape = config.get("data.shape").unwrap_or_default().to_owned();
            out.data_devices = (0..config.count("data"))
                .filter_map(|i| config.get(&format!("data.{i}.dev")).map(str::to_owned))
                .collect();
        }
        let data_label = config
            .and_then(|c| c.get("data.mountpoint"))
            .map(str::to_owned)
            .unwrap_or_else(|| out.data_root.clone());
        let serialdat_label = config
            .and_then(|c| c.get("serialdat.mountpoint"))
            .map(str::to_owned)
            .unwrap_or_else(|| out.slow_root.clone());
        let disk = crate::disktest::last();
        let data_fs = out
            .filesystems
            .iter()
            .find(|f| f.what == "data root")
            .cloned();
        out.glance = crate::glance::lines(&crate::glance::Facts {
            record_present: out.record_present,
            data_label: &data_label,
            data_shape: &out.data_shape,
            data_devices: &out.data_devices,
            serialdat_label: &serialdat_label,
            serialdat_class: &out.backing_class,
            serialdat_device: &out.backing_device,
            off_stripe: &out.off_stripe,
            off_stripe_health: &out.off_stripe_health,
            data_fs: data_fs.as_ref(),
            disk: disk.as_ref(),
        });
        out
    }

    fn try_mirror_identity(&self) -> Result<()> {
        let principals = self.auth.principals().context("listing principals")?;
        let keys = self.auth.keys().context("listing enrolled keys")?;
        let written_unix_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or_default();
        let doc = serde_json::json!({
            "written_unix_ms": written_unix_ms,
            "principals": principals,
            "keys": keys,
        });
        let bytes = serde_json::to_vec_pretty(&doc).context("encoding the identity mirror")?;
        let dir = self.slow.meta_dir()?;
        crate::durable::write_atomic_unframed(&dir, IDENTITY_MIRROR, &bytes)
    }

    /// The storage engine this process is running.
    pub fn engine(&self) -> crate::storage::Engine {
        self.engine
    }

    /// Where served repositories live.
    pub fn repos_root(&self) -> &Path {
        &self.repos_root
    }

    /// The identity table (the ACL model).
    /// **Who signed in and who changed something, and when** (2026-10-21) —
    /// the stamps `Access.ListPrincipals` answers beside each principal.
    pub fn activity(&self) -> &Arc<gunnar_core::PrincipalActivity> {
        &self.activity
    }

    pub fn auth(&self) -> &AuthTable {
        &self.auth
    }

    /// The repository catalog.
    pub fn catalog(&self) -> &RepoCatalog {
        &self.catalog
    }

    /// Per-repository policy documents (`refs/gunnar/policy`).
    ///
    /// The **one** home for branch protection and ref-name policy since
    /// `Policy.*` stopped answering `Unimplemented`. `gunnar-policy.json`
    /// beside a repository is still read as a fallback, and
    /// `transport::load_policy` says why and for how long.
    pub fn policies(&self) -> &gunnar_core::PolicyStore {
        &self.policies
    }

    /// Pull-mirror registrations (`refs/gunnar/mirrors`).
    ///
    /// The server's own registry, reachable over `Mirror.*`. Not the same thing
    /// as `gunnar-client`'s `mirrors.json`, which is the pull-mirror *client's*
    /// state on a host running `gunnar mirror`.
    pub fn mirrors(&self) -> &gunnar_core::MirrorStore {
        &self.mirrors
    }

    /// **The flow ledger**: what moved, per namespace, per UTC day.
    ///
    /// Handed out as an `Arc` because its holders outlive any one borrow — the
    /// transfer path keeps a clone for the length of a transfer, the flush tick
    /// keeps one for the length of the process — and because every holder must
    /// share the one set of live atomics. See the field.
    /// **Every account's audit log** — read only by each account's admins
    /// (`Enterprise.AuditLog`), written by the control plane's handlers.
    pub fn account_audit(&self) -> &Arc<gunnar_core::account_audit::AccountAuditLog> {
        &self.account_audit
    }

    /// **The name filter** — its lists and the accounts' review queues.
    pub fn names(&self) -> &gunnar_core::name_filter::NameFilterStore {
        &self.names
    }

    /// **People's encryption keys** (`Access.GetEncryptionKey` and friends, `Repos.RecipientKeys`).
    pub fn encryption_keys(&self) -> &gunnar_core::encryption_key::EncryptionKeys {
        &self.encryption_keys
    }

    /// **The generic server settings** (`Admin.GetSetting` / `Admin.SetSetting`).
    pub fn settings(&self) -> &gunnar_core::settings::SettingsStore {
        &self.settings
    }

    pub fn usage_ledger(&self) -> &Arc<crate::usage_ledger::UsageLedger> {
        &self.usage
    }

    /// The server's one set of counters (see the field).
    pub fn metrics(&self) -> &Arc<gunnar_core::observe::ServerMetrics> {
        &self.metrics
    }

    pub fn flow(&self) -> &Arc<gunnar_core::FlowLedger> {
        &self.flow
    }

    /// **Pin the stores for a snapshot streamed from a cut** (R5.1-A, 2026-10-07): until the
    /// guard drops, [`ServerStores::wait_unpinned`] holds back a gc, a repository store's removal
    /// and an LFS object's removal — the objects reachable from the cut's tips, the stores it
    /// names and the LFS objects it lists stay where the stream will read them. Take it BEFORE the
    /// log's freeze; the ones who wait take it before their own mutation guard, so a freeze can
    /// never wait on a thread that waits on a pin.
    pub fn snapshot_pin(&self) -> SnapshotPin<'_> {
        *self
            .snapshot_pins
            .count
            .lock()
            .unwrap_or_else(|p| p.into_inner()) += 1;
        SnapshotPin {
            pins: &self.snapshot_pins,
        }
    }

    /// **Wait while a snapshot streams from a cut** ([`ServerStores::snapshot_pin`]) — at most
    /// [`SNAPSHOT_PIN_WAIT`], after which the caller goes on and that snapshot fails at the piece
    /// that needed what was removed (the twin asks again; it never installs half). `what` names the
    /// caller in the line said while it waits. Called BEFORE any mutation guard.
    pub fn wait_unpinned(&self, what: &str) {
        let pins = &self.snapshot_pins;
        let mut n = pins.count.lock().unwrap_or_else(|p| p.into_inner());
        if *n == 0 {
            return;
        }
        tracing::info!(target: "gunnar::store", what, snapshots = *n, "a snapshot is streaming from its cut: {what} waits for it");
        let deadline = std::time::Instant::now() + SNAPSHOT_PIN_WAIT;
        while *n > 0 {
            let left = deadline.saturating_duration_since(std::time::Instant::now());
            if left.is_zero() {
                tracing::warn!(target: "gunnar::store", what, "a snapshot still streams after {} s: {what} goes on; that snapshot fails and is asked for again", SNAPSHOT_PIN_WAIT.as_secs());
                return;
            }
            n = pins
                .cv
                .wait_timeout(n, left)
                .unwrap_or_else(|p| p.into_inner())
                .0;
        }
    }

    /// The catalog and the identity table, as one store (the certificate-identity plan).
    /// The transaction log — what the twin fetches and what a restore replays.
    pub fn txlog(&self) -> &Arc<gunnar_core::txlog::Log> {
        &self.txlog
    }

    /// **Is this box a twin right now?** Set at open from `twin.json`, and by
    /// [`ServerStores::mark_twin`] the moment `Bootstrap.BecomeTwin` confers
    /// the role on a running box. NOT decided once, whatever this comment used
    /// to say.
    pub fn is_twin(&self) -> bool {
        self.twin.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// **The shared twin flag**, for the front door — the same `Arc`
    /// [`ServerStores::mark_twin`] writes, so a box conferred the twin role
    /// over the wire refuses git from that instant instead of from the next
    /// restart. See `gunnar_ssh::Control::with_twin_flag`.
    pub fn twin_flag(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.twin)
    }

    /// From now on this box is a twin (`crate::twin::become_twin`).
    pub fn mark_twin(&self) {
        self.twin.store(true, std::sync::atomic::Ordering::SeqCst);
        // No derived memo on a twin's disk from the next store open on (C2), no exploded read
        // cache and a resolver narrow enough for the box (C1 + the twin memory bound). This is
        // the runtime half that made the five `korp.env.ZNIPPY_*` tokens retirable, so a twin
        // needs nothing on its command line.
        self.engine.hold_no_graph_memo();
        self.engine.hold_twin_memory_bound();
        // A setter governs the NEXT open, so the stores this box already holds — it came up a
        // primary, which is what one medium means — are swept by hand.
        let freed = self.engine.bound_open_stores_to_twin();
        if freed > 0 {
            tracing::info!(
                target: gunnar_core::observe::TARGET_TWIN,
                bytes = freed,
                "unlinked the exploded read cache of the repositories this box already held: a twin holds none"
            );
        }
    }

    /// The sealed secret store over the UNLOGGED half: what a twin reads its
    /// replicated host key through (a twin frames nothing).
    pub fn secrets_unframed(
        &self,
    ) -> Result<gunnar_core::SealedSecretStore<std::sync::Arc<dyn gunnar_store::SecretStore>>> {
        Ok(gunnar_core::SealedSecretStore::new(
            Arc::clone(&self.raw_secrets),
            self.kek(),
        ))
    }

    fn kek(&self) -> gunnar_core::Kek {
        self.kek
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    /// **Hold `bytes` as this process's KEK** (S9): what a follower does with a `Kek` frame
    /// once the key file is durable, so the sealed rows it holds open from now on.
    pub fn adopt_kek(&self, bytes: &[u8]) -> Result<()> {
        let key: [u8; 32] = bytes
            .try_into()
            .map_err(|_| anyhow::anyhow!("a KEK carries {} bytes, not 32", bytes.len()))?;
        *self
            .kek
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = gunnar_core::Kek::from_bytes(key);
        Ok(())
    }

    /// The KEK this process holds, as bytes — a snapshot's `Kek` item when no key file exists
    /// (the key was handed in by the environment).
    pub fn kek_bytes(&self) -> Vec<u8> {
        hex::decode(self.kek().to_hex()).expect("a KEK's hex decodes")
    }

    /// **The control store with no log in front of it** — the twin applies
    /// the primary's `ControlBlob` / `ControlDoc` frames here, and nothing
    /// else writes here.
    pub fn raw_control_store(&self) -> &Arc<dyn ControlStore> {
        &self.raw_control_store
    }

    /// **The secret store with no log in front of it** — the twin's
    /// `Secret` frames land here as the sealed bytes they are.
    pub fn raw_secrets(&self) -> &Arc<dyn gunnar_store::SecretStore> {
        &self.raw_secrets
    }

    pub fn control_store(&self) -> &Arc<dyn ControlStore> {
        &self.control_store
    }

    /// The redb handle under the control store — see the field.
    pub fn control_db(&self) -> Arc<gunnar_core::ControlDb> {
        Arc::clone(&self.control_db)
    }

    /// Are the repository stores closed for a growth?
    pub fn suspended(&self) -> bool {
        self.suspended.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// **Close every repository store for the runtime growth** (`crate::grow`,
    /// `DATA-SET-GROWTH-FLOW.md` G6): from here until [`Self::resume_repos`]
    /// `repo_store` refuses with `why`, the handle cache is emptied and the
    /// engines release what they cached — the packs' mmaps, the redb tails,
    /// the descriptors that would otherwise pin `/data`. A store still held
    /// by an in-flight request (there are none: the drain finished first)
    /// drops when that request returns.
    pub fn suspend_repos(&self, why: &str) {
        self.suspended
            .store(true, std::sync::atomic::Ordering::SeqCst);
        let dropped = {
            let mut cache = self.cache();
            let n = cache.entries.len();
            cache.entries.clear();
            n
        };
        crate::storage::shutdown();
        tracing::warn!(
            target: "gunnar::store",
            event = "store.suspended",
            dropped,
            why,
            "every repository store is closed and opens are refused: {why}"
        );
    }

    /// The stores may open again (lazily, on the next request), after the
    /// growth mounted `/data` back.
    pub fn resume_repos(&self) {
        // The data set was unmounted and is back, perhaps grown: every reading
        // the usage ledger holds is re-taken.
        self.usage.mark_all();
        if self
            .suspended
            .swap(false, std::sync::atomic::Ordering::SeqCst)
        {
            tracing::info!(target: "gunnar::store", event = "store.resumed", "repository stores open again on demand");
        }
    }

    /// Where the KEK and the legacy Arrow secret log live. See [`CONTROL_DIR`].
    ///
    /// `control()` used to be here — the one accessor on this type that opened
    /// a repository, and the reason the server needed a storage engine to boot.
    /// There is no control repository any more, so there is no accessor for one.
    pub fn control_dir(&self) -> &Path {
        &self.control_dir
    }

    /// **The server's secret store, sealed.**
    ///
    /// Everything on the boot path must go through this rather than through the
    /// raw handle: the raw handle writes plaintext, and the SSH host key is the
    /// one secret whose whole purpose is to be persisted. The decorator is the
    /// only thing standing between a private key and a file designed to be
    /// copied off the machine — and now that the store is a table in
    /// `control.redb`, "the file designed to be copied off the machine" is the
    /// same file the catalog is in.
    ///
    /// Still `Result`, and deliberately: the signature is what every caller on
    /// the boot path is written against, and the store behind it may yet be one
    /// that can fail to open (`VaultSecretStore` is still noted-not-built). It
    /// no longer opens anything.
    /// # And what it does with a value that is not sealed
    ///
    /// It refuses it, unless [`Self::legacy_secrets`] says a pre-sealing
    /// `secrets.arrow` is on this disk. That is not belt-and-braces: reading an
    /// unsealed value through is a way past the seal that needs no KEK at all,
    /// so anything that can write `control.redb` — a restored backup, a tampered
    /// cold-tier archive, a replica's copied file — could put its own SSH host
    /// key in the slot and gunnar would serve it. Only the machine that has the
    /// old log has a legitimate unsealed value to read, and only that machine
    /// gets the policy that reads one.
    ///
    /// [`Self::legacy_secrets`]: ServerStores::legacy_secrets
    pub fn secrets(
        &self,
    ) -> Result<gunnar_core::SealedSecretStore<std::sync::Arc<dyn gunnar_store::SecretStore>>> {
        let inner = Arc::clone(&self.secrets);
        Ok(if self.legacy_secrets {
            gunnar_core::SealedSecretStore::migrating(inner, self.kek())
        } else {
            gunnar_core::SealedSecretStore::new(inner, self.kek())
        })
    }

    /// **The transparency log's signing key**, minting one on first use.
    ///
    /// Its own key and not the issuer's: the issuer says who somebody *is*, and
    /// this says what the log *held*. One key for both would mean compromising
    /// the log's signer also mints administrators, and would make revoking one
    /// job impossible without revoking the other.
    ///
    /// Atomic across racing callers, exactly as the issuer key and the host key
    /// are: two heads issued at once produce one key, not two.
    pub fn transparency_key(&self) -> Result<gunnar_core::ssh_key::PrivateKey> {
        use gunnar_store::SecretStoreExt as _;
        let secrets = self.secrets()?;
        let bytes = secrets
            .get_or_create_with(gunnar_store::TRANSPARENCY_LOG_KEY, || {
                tracing::info!(
                    key = gunnar_store::TRANSPARENCY_LOG_KEY,
                    "no transparency-log signing key stored; minting one. It is sealed under the \
                     KEK, and its public half is what a verifier pins to check a signed tree head"
                );
                gunnar_core::mint_ed25519()
                    .map_err(|err| gunnar_store::Error::backend(format!("{err}")))
            })
            .context("reading the transparency log's signing key")?;
        let pem = String::from_utf8(bytes)
            .context("the stored transparency-log key is not utf-8 OpenSSH PEM")?;
        gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .context("parsing the transparency log's signing key")
    }

    /// The transparency log over this server's push attestations.
    pub fn transparency(&self) -> crate::attest::TransparencyLog {
        crate::attest::TransparencyLog::under(&self.root)
    }

    /// The transparency log over this server's **build** records.
    ///
    /// The same machinery over a different file: a build record and a push
    /// attestation are different shapes with different readers, and mixing them
    /// into one file would break every reader of the other.
    pub fn build_log(&self) -> crate::attest::TransparencyLog {
        crate::attest::TransparencyLog::over(
            self.root.join(crate::attest::BUILD_LOG),
            self.root.join(crate::attest::BUILD_HEADS),
        )
    }

    /// The transparency log over this server's **forge** events.
    ///
    /// Issues, reviews and merges, each a row signed by the principal that did
    /// it — so an approval is evidence in the same sense a signed commit is,
    /// and not the server's word for it.
    pub fn forge_log(&self) -> crate::attest::TransparencyLog {
        crate::attest::TransparencyLog::over(
            self.root.join(crate::attest::FORGE_LOG),
            self.root.join(crate::attest::FORGE_HEADS),
        )
    }

    /// The transparency log over this server's **group-membership** changes.
    ///
    /// The same machinery over a different file: each line is an add or remove
    /// signed by the acting admin's own key, so who was in a group at any time
    /// is evidence a third party can check and not the server's word. The live
    /// membership table ([`gunnar_core::AuthTable`]) stays the authority the SSH
    /// listener resolves grants through; this is the durable, signed record of
    /// how it came to say what it says.
    pub fn membership_log(&self) -> crate::attest::TransparencyLog {
        crate::attest::TransparencyLog::over(
            self.root.join(crate::attest::MEMBERSHIP_LOG),
            self.root.join(crate::attest::MEMBERSHIP_HEADS),
        )
    }

    /// **The builder's provenance signing key**, minting one on first use.
    ///
    /// A third key beside the issuer's and the log's, because it is a third
    /// job — and because a runner on its own host needs a key of its own that
    /// revoking the log's signer does not take with it.
    pub fn provenance_key(&self) -> Result<gunnar_core::ssh_key::PrivateKey> {
        use gunnar_store::SecretStoreExt as _;
        let secrets = self.secrets()?;
        let bytes = secrets
            .get_or_create_with(gunnar_store::PROVENANCE_KEY, || {
                tracing::info!(
                    key = gunnar_store::PROVENANCE_KEY,
                    "no provenance signing key stored; minting one. Its public half is what a \
                     verifier pins to check a build record"
                );
                gunnar_core::mint_ed25519()
                    .map_err(|err| gunnar_store::Error::backend(format!("{err}")))
            })
            .context("reading the provenance signing key")?;
        let pem = String::from_utf8(bytes)
            .context("the stored provenance key is not utf-8 OpenSSH PEM")?;
        gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .context("parsing the provenance signing key")
    }

    /// **The outbound webhook signing key**, minting one on first use.
    ///
    /// A fourth key beside the issuer's, the log's and the builder's, because a
    /// delivery signature is a fourth job: it says "this event body came from
    /// this server". Its PUBLIC half is what a webhook receiver pins and checks
    /// each delivery against with `ssh-keygen -Y verify` — no shared secret, so
    /// a leaked delivery URL is not a leaked signing key. Sealed under the KEK
    /// and atomic across racing callers, exactly as the keys above are.
    pub fn webhook_key(&self) -> Result<gunnar_core::ssh_key::PrivateKey> {
        use gunnar_store::SecretStoreExt as _;
        let secrets = self.secrets()?;
        let bytes = secrets
            .get_or_create_with(gunnar_store::WEBHOOK_KEY, || {
                tracing::info!(
                    key = gunnar_store::WEBHOOK_KEY,
                    "no webhook signing key stored; minting one. Its public half is what a \
                     receiver pins to verify a delivery"
                );
                gunnar_core::mint_ed25519()
                    .map_err(|err| gunnar_store::Error::backend(format!("{err}")))
            })
            .context("reading the webhook signing key")?;
        let pem =
            String::from_utf8(bytes).context("the stored webhook key is not utf-8 OpenSSH PEM")?;
        gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .context("parsing the webhook signing key")
    }

    /// **The forge-attestation key**, minting one on first use (GHUNT-19):
    /// what gunnar signs a person's forge event with when it came from their
    /// stepped-up session ([`gunnar_store::FORGE_ATTEST_KEY`]). Sealed under the
    /// KEK and atomic across racing callers, exactly as the keys above are.
    pub fn forge_attest_key(&self) -> Result<gunnar_core::ssh_key::PrivateKey> {
        use gunnar_store::SecretStoreExt as _;
        let secrets = self.secrets()?;
        let bytes = secrets
            .get_or_create_with(gunnar_store::FORGE_ATTEST_KEY, || {
                tracing::info!(
                    key = gunnar_store::FORGE_ATTEST_KEY,
                    "no forge-attestation key stored; minting one. It signs a person's forge \
                     event made from a session that stepped up"
                );
                gunnar_core::mint_ed25519()
                    .map_err(|err| gunnar_store::Error::backend(format!("{err}")))
            })
            .context("reading the forge-attestation key")?;
        let pem = String::from_utf8(bytes)
            .context("the stored forge-attestation key is not utf-8 OpenSSH PEM")?;
        gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .context("parsing the forge-attestation key")
    }

    /// The forge-attestation key's PUBLIC half, if one was ever minted — read
    /// without minting, so a reader (a twin included) never writes a secret.
    pub fn forge_attest_public(&self) -> Result<Option<gunnar_core::ssh_key::PublicKey>> {
        use gunnar_store::SecretStore as _;
        let secrets = self.secrets()?;
        let Some(bytes) = secrets
            .get(gunnar_store::FORGE_ATTEST_KEY)
            .context("reading the forge-attestation key")?
        else {
            return Ok(None);
        };
        let pem = String::from_utf8(bytes)
            .context("the stored forge-attestation key is not utf-8 OpenSSH PEM")?;
        Ok(Some(
            gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
                .context("parsing the forge-attestation key")?
                .public_key()
                .clone(),
        ))
    }

    /// **The actor-ticket signing key**, minting one on first use.
    ///
    /// A fifth key beside the issuer's, the log's, the builder's and the
    /// webhook's, and a fifth job: it signs the appliance's word that one
    /// human may act for one tenant. Sealed under the KEK and atomic across
    /// racing callers, exactly as the four above are.
    ///
    /// # Why not the SSH host key, which is right there
    ///
    /// Because the host key's signatures always travel inside something that
    /// separates domains for them — the SSH transport's own framing, or the
    /// `sshsig` envelope with its namespace — and a raw Ed25519 signature over
    /// a JSON document travels inside nothing. One key signing in two formats,
    /// one of which has no domain separator at the signature layer, is a
    /// cross-protocol confusion that costs nothing today and is not
    /// recoverable on the day somebody finds bytes that parse as both. A key
    /// per job also keeps a compromise of one a compromise of one, which is
    /// the sentence the other four are each here for.
    ///
    /// # Why this costs no installer pin, which was the alternative
    ///
    /// It is minted AT RUNTIME into the secret store this box already has.
    /// The design that was weighed against it — ship the key, or a
    /// configuration naming it, in the image — would have put a new value into
    /// the two manifests that pin the installer, and `gunnar-cli`'s own note
    /// on that pin is the reason it was not taken: a key that can only be
    /// rolled by re-cutting an ISO is a key nobody rolls. Nothing here reaches
    /// the image; the first `Billing.MintActorTicket` on a fresh appliance
    /// mints it, and `Billing.ApplianceIdentity` hands the operator the public
    /// half to paste into their configuration.
    ///
    /// # Why it is raw Ed25519 and not `sshsig`
    ///
    /// Because the VERIFIER is `monetize_embed::ticket::verify_ticket`, which
    /// holds an `ed25519_dalek::VerifyingKey` and checks the canonical bytes
    /// that crate defines. gunnar writes no second canonical form and no
    /// second envelope — the stored key is still an ordinary OpenSSH ed25519
    /// private key, minted by the same [`gunnar_core::mint_ed25519`] as its
    /// four siblings, and only the SIGNATURE format differs.
    /// **The named sealed slot `name`** (2026-10-09): `(owner principal id,
    /// value)`, or `None`. Stored as the secret `slot/<name>` through
    /// [`Self::secrets`] — sealed under the KEK with the key as its AAD, one
    /// `Secret` frame — so the twin copies it, a snapshot carries it and a
    /// recharge restores it with no code of its own. The ACL is the gRPC
    /// service's; this reads and writes.
    #[cfg(feature = "metered")]
    pub fn sealed_slot_get(&self, name: &str) -> Result<Option<(String, Vec<u8>)>> {
        use gunnar_store::SecretStore as _;
        let raw = self
            .secrets()?
            .get(&sealed_slot::key(name))
            .context("reading a sealed slot")?;
        raw.map(|b| sealed_slot::decode(&b)).transpose()
    }

    /// **Store `value` in the slot `name`, owned by `owner`.** Answers whether
    /// it was created (it did not exist).
    #[cfg(feature = "metered")]
    pub fn sealed_slot_put(&self, name: &str, owner: &str, value: &[u8]) -> Result<bool> {
        use gunnar_store::SecretStore as _;
        let secrets = self.secrets()?;
        let key = sealed_slot::key(name);
        let created = secrets
            .get(&key)
            .context("reading a sealed slot")?
            .is_none();
        secrets
            .put(&key, &sealed_slot::encode(owner, value))
            .context("writing a sealed slot")?;
        Ok(created)
    }

    #[cfg(feature = "metered")]
    pub fn actor_ticket_key(&self) -> Result<crate::actor_ticket::ActorTicketKey> {
        use gunnar_store::{SecretStore as _, SecretStoreExt as _};
        let secrets = self.secrets()?;
        // RCH-TK: a re-laid main that awaits its twin, or is being restored, gets the DEAD
        // main's key back with the twin's sealed secrets — the key the ONE box still holds for
        // this site. Minting one first would sign tickets nobody accepts.
        let root = self.root();
        if (root.join(crate::twin::RECHARGE_REL).exists() || crate::twin::is_restoring(root))
            && secrets
                .get(gunnar_store::ACTOR_TICKET_KEY)
                .context("reading the actor-ticket signing key")?
                .is_none()
        {
            anyhow::bail!("this box awaits its twin's restore (a re-laid main): the actor-ticket key comes back with it, none is minted here");
        }
        let bytes = secrets
            .get_or_create_with(gunnar_store::ACTOR_TICKET_KEY, || {
                tracing::info!(
                    key = gunnar_store::ACTOR_TICKET_KEY,
                    "no actor-ticket signing key stored; minting one. Its public half is what \
                     the other side of the seam pins to check what this appliance signed"
                );
                gunnar_core::mint_ed25519()
                    .map_err(|err| gunnar_store::Error::backend(format!("{err}")))
            })
            .context("reading the actor-ticket signing key")?;
        let pem = String::from_utf8(bytes)
            .context("the stored actor-ticket key is not utf-8 OpenSSH PEM")?;
        let private = gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .context("parsing the actor-ticket signing key")?;
        crate::actor_ticket::ActorTicketKey::from_openssh(&private)
    }

    /// **The server's certificate issuer** (the certificate-identity plan).
    ///
    /// Built from the two halves that must never be one: the control store,
    /// which holds the issuer *certificate* and the trusted-CA list, and the
    /// **sealed** secret store, which holds the private key. It is
    /// [`Self::secrets`] that is handed over and never the raw handle — the raw
    /// handle writes plaintext, and this is the one key on the machine whose
    /// loss means an attacker issues themselves an administrator rather than
    /// having to break anything.
    ///
    /// The two halves are now two **tables** rather than two files, and the
    /// boundary that protects the key is the seal rather than the file: the KEK
    /// is in `secrets.kek` beside the database and not in it, so a copied
    /// `control.redb` carries a certificate it cannot sign with.
    ///
    /// It no longer needs a storage engine. Neither does
    /// [`Self::pki_readonly`], which is why both are here and only the
    /// difference between them is about what they may write.
    pub fn pki(&self) -> Result<gunnar_core::Pki> {
        let secrets: std::sync::Arc<dyn gunnar_store::SecretStore> =
            std::sync::Arc::new(self.secrets()?);
        Ok(gunnar_core::Pki::new(
            Arc::clone(&self.control_store),
            secrets,
        )?)
    }

    /// **The issuer certificate and the trusted-CA list, and nothing that can
    /// sign.**
    ///
    /// `gunnar issuer show` is a control-plane read. The returned handle refuses
    /// every secret-store call **by construction**, so it cannot mint, import or
    /// issue — see [`gunnar_core::Pki::read_only`]. That is now the only thing
    /// separating it from [`Self::pki`]: since the secrets moved into
    /// `control.redb`, "without an engine" describes both, and the distinction
    /// this method carries is a capability one rather than a dependency one.
    pub fn pki_readonly(&self) -> Result<gunnar_core::Pki> {
        Ok(gunnar_core::Pki::read_only(Arc::clone(
            &self.control_store,
        ))?)
    }

    /// The identity table and catalog together, as `gunnar-ssh` wants them.
    ///
    /// Handed the shared write-gate so the front door refuses a client push on
    /// a standby. Every `Control` this returns clones the SAME gate, so SSH and
    /// smart-HTTP freeze together the instant a role change flips it.
    /// **Install the plan-gate chain** a mirror sync is held to — the front
    /// doors' own, in their order. Replaces whatever was installed: the chain
    /// is built once per serve, and the newest build is the one in force.
    pub fn install_plan_gates(&self, gates: Vec<Arc<dyn gunnar_ssh::PlanGate>>) {
        *self
            .plan_gates
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = gates;
    }

    /// **Install the data set's space ledger** — the disk gate's
    /// ([`gunnar_ssh::DiskHeadroomGate::ledger`]). Replaces any earlier one.
    pub fn install_space_ledger(&self, ledger: Arc<gunnar_ssh::SpaceLedger>) {
        *self
            .space
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(ledger);
    }

    /// The installed space ledger, if any.
    pub fn space_ledger(&self) -> Option<Arc<gunnar_ssh::SpaceLedger>> {
        self.space
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    /// The installed chain, in the order it is asked. Empty: no ceiling.
    pub fn plan_gates(&self) -> Vec<Arc<dyn gunnar_ssh::PlanGate>> {
        self.plan_gates
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    pub fn ssh_control(&self) -> Control {
        Control::new(self.auth.clone(), self.catalog.clone())
            .with_encryption_keys(self.encryption_keys.clone())
            .with_write_gate(Arc::clone(&self.writes_frozen))
            .with_browse_flag(Arc::clone(&self.browse))
            // **And the twin flag** (2026-09-21, lane T4): a twin serves no git
            // at all, and the role now arrives over the wire on a box that is
            // already listening. One `Arc`, so the door cannot be a restart
            // behind the role.
            .with_twin_flag(Arc::clone(&self.twin))
            // The owner's enterprise list (2026-10-05): a first push's
            // repository name meets the name filter, and its creation is in
            // its account's audit log.
            .with_enterprise(self.names.clone(), Arc::clone(&self.account_audit))
    }

    /// **The shared browse flag.** `true` = this box is a browse replica: a
    /// frozen standby that serves anonymous reads and refuses every
    /// control-plane mutator and its own promotion. Same one value as
    /// [`Self::writes_frozen`]'s sibling (one implementation).
    pub fn browse(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.browse)
    }

    /// **The shared write-gate.** `true` = this box is a frozen standby.
    ///
    /// The replication slot writes it on a role change; the SSH/HTTP front
    /// doors read it on every push. One value (one implementation).
    pub fn writes_frozen(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.writes_frozen)
    }

    /// **How many repository stores this process has opened so far** — see the
    /// `store_opens` field. Monotonic, so a caller measures a delta across the
    /// operation it is pricing. The measurement behind the red: a plain
    /// `Mirror.ReplicationStatus` must leave this unchanged.
    pub fn store_opens(&self) -> u64 {
        self.store_opens.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Where a repository's git directory is, checked against symlink escapes.
    ///
    /// The string sandbox in [`gunnar_core::resolve_repo_path`] cannot see a
    /// symlink; that check lives in [`RepoRef::path_under`], and this is the
    /// one place gunnar calls it.
    pub fn repo_dir(&self, path: &RepoRef) -> Result<PathBuf> {
        path.path_under(&self.repos_root)
            .with_context(|| format!("resolving the directory for {path}"))
    }

    /// The handle cache, recovering from a poisoned lock.
    ///
    /// This lock is taken on every repository lookup on every request, and it
    /// used to be three `.expect()`s. One panic anywhere under it poisoned the
    /// mutex and every subsequent request panicked: a whole server lost to one
    /// bad repository. What the lock protects is a *cache*: every entry in it is
    /// a complete, already-opened handle, so a panicking thread cannot leave it
    /// half-written, and recovering is strictly better than propagating.
    fn cache(&self) -> MutexGuard<'_, HandleCache> {
        self.open
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    /// **The bound the handle cache is currently holding itself to.**
    ///
    /// The companion to [`Self::open_handles`], which answers how many are held
    /// but not what against. `reclaim_under_pressure` MOVES this number, so a
    /// caller — or a test — that reads only the count cannot tell a cache that
    /// evicted from a cache that was cut.
    pub fn open_repo_capacity(&self) -> usize {
        self.cache().capacity
    }

    /// **Lower (or raise) the handle bound and evict down to it now.** Returns
    /// how many handles were dropped.
    ///
    /// The same call [`Self::reclaim_under_pressure`] makes, exposed because
    /// that one is reachable only from a `/proc/meminfo` reading below
    /// [`RECLAIM_FLOOR_PERCENT`] — a state a test cannot put a shared build box
    /// into, and the reason the reclaim path is the one half of the bound that
    /// `gunnar-cli/tests/handle_cache.rs` could not reach. Eviction on *insert*
    /// is covered there and is not duplicated here (one implementation); eviction on
    /// *pressure* had never run outside the appliance it was written for.
    pub fn set_open_repo_capacity(&self, capacity: usize) -> usize {
        self.cache().set_capacity(capacity)
    }

    /// The stores for one catalogued repository, opening it if this is the
    /// first request for it.
    ///
    /// The returned `Arc` is what keeps the handle alive: while a caller holds
    /// one, the evictor will not drop that repository from the cache, so an
    /// in-flight clone cannot have its store lock pulled out from under it.
    pub fn repo_store(&self, repo: &Repo) -> Result<Arc<OpenRepo>> {
        let path = repo.path();
        let key = path.key();
        if self.suspended() {
            anyhow::bail!(
                "the data set is off-line while it is being grown; the repository {} cannot be \
                 opened until /data is mounted again — retry in a few minutes",
                key
            );
        }
        if let Some(store) = self.cache().touch(&key) {
            return Ok(store);
        }
        // A miss is about to open a store, which is the one moment the cache
        // grows. Check the box before it does; see `reclaim_under_pressure`.
        self.reclaim_under_pressure();

        let dir = self.repo_dir(&path)?;
        if let Some(parent) = dir.parent() {
            std::fs::create_dir_all(parent).with_context(|| {
                format!("creating the namespace directory {}", parent.display())
            })?;
        }
        // The engine opens it. Which engine that is, is a fact about startup
        // rather than about this file.
        // The one site a store is actually opened. Count it here — after the
        // cache miss and before anything can early-return — so `store_opens`
        // measures engine opens (the refold cost) and nothing a cache
        // hit avoided. See the field.
        self.store_opens
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        // An open store is one whose bytes may move: the usage ledger
        // re-measures it while it is open and once after it closes.
        self.usage.mark_repo(&repo.namespace, &repo.name);
        let store = self
            .engine
            .open_repo(&dir, repo.hash_kind.into())
            .with_context(|| format!("opening the repository at {}", dir.display()))?
            .logged(
                Arc::clone(&self.txlog),
                &repo.key(),
                &repo.hash_kind.to_string(),
            );
        Self::point_head(&store, repo)?;
        // The hash marker, written once. `--adopt` is the only reader; a
        // store from before 2026-08-30 gets its marker on its first open
        // after the upgrade, from the catalog row it has always been served
        // under.
        let marker = dir.join(Self::HASH_MARKER);
        if !marker.exists() {
            if let Err(err) = std::fs::write(&marker, format!("{}\n", repo.hash_kind)) {
                tracing::warn!(
                    target: "gunnar::store",
                    repo = %repo.key(),
                    marker = %marker.display(),
                    error = %err,
                    "the hash marker could not be written; --adopt will refuse this store"
                );
            }
        }

        // **The account's quota, applied at the one moment it can be.** The
        // store opens on the server's default and is moved onto the plan here,
        // so an admin who set a quota while the repository was closed does not
        // have to set it again. A namespace with no quota keeps the default,
        // which is what `None` means and why the call is skipped rather than
        // passed a zero.
        self.apply_explode_quota(&repo.namespace, &dir);

        // Another thread may have won the race while this one was on disk. Its
        // handle holds the lock; ours is about to be dropped, releasing it. So
        // keep the winner rather than overwriting it.
        Ok(self.cache().insert(key, store))
    }

    /// Move every open store under `dir` onto the namespace's quota.
    ///
    /// Called when a store opens and when an admin changes the plan. A failure
    /// to read the catalog is logged and the default stands: a quota that
    /// cannot be read must not stop a repository serving, because the worst it
    /// costs is a cache that is the wrong size and the best it costs is an
    /// outage.
    pub fn apply_explode_quota(&self, namespace: &str, dir: &Path) {
        match self.catalog().namespace_explode_budget(namespace) {
            Ok(Some(bytes)) => {
                let n = self.engine.set_explode_budget(dir, bytes);
                tracing::debug!(
                    target: "gunnar::store",
                    namespace,
                    bytes,
                    stores = n,
                    "the account's explode budget was applied"
                );
            }
            Ok(None) => {}
            Err(err) => tracing::warn!(
                target: "gunnar::store",
                namespace,
                error = %err,
                "the account's explode budget could not be read; the server default stands"
            ),
        }
    }

    /// **Shrink the handle cache when the box is short of memory.**
    ///
    /// # The hole this closes — MEASURED on the 2 GB appliance, 2026-08-30
    ///
    /// On a 2 GB appliance the box collapsed at ~1200 repositories:
    /// `MemAvailable` fell under the indexer watchdog's floor (399 MB) and
    /// every store then spilled caches it did not hold — `scanned_rows_dropped=0`
    /// on every push — because the memory was in the **other 255 handles in the
    /// LRU**, which the watchdog cannot reach. Before the console fix that
    /// ended in a hard wedge: eight transfers in flight, no progress for 41
    /// minutes, the control plane still answering. The handle cache was bounded
    /// by descriptors and nothing else, so it never gave anything back.
    ///
    /// [`crate::budget`] sizes the cache against memory at start. This is the
    /// same rule applied *again*, from the memory the box actually has now:
    /// below [`RECLAIM_FLOOR_PERCENT`] of `MemTotal` available, the bound is
    /// halved and the least recently used handles are dropped, freeing their
    /// redb page caches and their five descriptors each. It is the eviction the
    /// watchdog could not do, done by the only thing that owns the handles.
    ///
    /// **Rate-limited to one probe per [`RECLAIM_PROBE_INTERVAL`]**, and taken
    /// only on a cache *miss*, which is the one moment the cache grows. A
    /// `/proc/meminfo` read on every request would be a syscall on the hot
    /// path; on a miss it is noise beside opening a store.
    ///
    /// It never shrinks below [`crate::limits::MIN_OPEN_REPOS`]: a server that
    /// can keep no repository open re-opens on every request, and a box that
    /// low is the WARN row's job, not the cache's.
    fn reclaim_under_pressure(&self) {
        let now = std::time::Instant::now();
        {
            let mut probed = self
                .last_reclaim_probe
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            match *probed {
                Some(last) if now.duration_since(last) < RECLAIM_PROBE_INTERVAL => return,
                _ => *probed = Some(now),
            }
        }
        let Some((total, available)) = crate::dashboard::probe::memory() else {
            return;
        };
        if total == 0 || available >= total / 100 * RECLAIM_FLOOR_PERCENT {
            return;
        }
        let mut cache = self.cache();
        let before = cache.capacity;
        let want = (before / 2).max(crate::limits::MIN_OPEN_REPOS);
        if want >= before {
            return;
        }
        let dropped = cache.set_capacity(want);
        let held = cache.entries.len();
        drop(cache);
        tracing::warn!(
            target: "gunnar::limits",
            event = "limits.handles.reclaimed",
            mem_available = available,
            mem_total = total,
            was = before,
            now = want,
            dropped,
            open = held,
            "memory is short ({} of {} available): the open-repository cache is cut from {} to \
             {} handles and {} were dropped. Each one held five descriptors and its page caches; \
             a repository touched again simply re-opens.",
            crate::budget::human(available),
            crate::budget::human(total),
            before,
            want,
            dropped
        );
    }

    /// Make an open repository's `HEAD` name the default branch its catalog
    /// entry carries.
    ///
    /// **The only place the server projects the catalog onto a repository's
    /// `HEAD`** (one implementation), and it is called from exactly two: [`Self::repo_store`]
    /// on the open path, so create and every transfer after a restart carry it,
    /// and `ControlPlane::set_default_branch`, which has to re-run it because
    /// the handle it needs is already in the cache and `repo_store` returns
    /// early for those.
    ///
    /// A move is logged at info. It is a change to what a clone checks out and
    /// it happens without anyone asking on the open path, so it must be visible
    /// in the log of the repository that had it done to it.
    pub fn point_head(store: &OpenRepo, repo: &Repo) -> Result<()> {
        // Through the UNLOGGED store: HEAD's default is derived from the
        // catalog row, which is itself a logged document, and every box —
        // primary, twin, a restore — derives it the same way on open. Logging
        // it here would put a frame in a twin's log that the primary never
        // numbered, and `append_at` would refuse the primary's next one.
        let moved = gunnar_core::point_head_at(store.unlogged(), &repo.default_branch)
            .with_context(|| format!("pointing {}'s HEAD at its default branch", repo.key()))?;
        if moved {
            tracing::info!(
                repo = %repo.key(),
                default_branch = %repo.default_branch,
                "HEAD now names the repository's default branch; it did not before, so this \
                 repository was not clonable if that branch was the only one with commits"
            );
        }
        Ok(())
    }

    /// [`Self::point_head`] for a repository this server has catalogued,
    /// opening its store first. Used after the catalog's default branch moves.
    pub fn repoint_head(&self, repo: &Repo) -> Result<()> {
        let store = self.repo_store(repo)?;
        Self::point_head(&store, repo)
    }

    /// Adopt a pushed branch as the default branch of a repository whose `HEAD`
    /// still resolves to nothing, and report the name if one was adopted.
    ///
    /// # Why a push is allowed to change this
    ///
    /// `repo create` gives every repository `refs/heads/main`, and a donor
    /// whose branch is `master` therefore lands in a repository whose `HEAD`
    /// names a branch that does not exist and never will. That repository
    /// **cannot be cloned**: `HEAD` has no object, so it is not advertised, and
    /// stock `git clone` writes an empty worktree, prints *"remote HEAD refers
    /// to nonexistent ref"* and exits **0**. Nothing between the successful push
    /// and the empty checkout says a word about the default branch, which is
    /// what cost the roundtrip harness a day on 2026-08-06.
    ///
    /// So the first branch to arrive in a repository with an unborn `HEAD` is
    /// adopted, which is what every hosted forge does and what makes `push`
    /// alone produce something clonable. It is deliberately narrow:
    ///
    /// * it fires **only** when `HEAD` resolves to nothing *after* the push, so
    ///   the only repositories it can touch are the ones that are unclonable as
    ///   they stand — if the configured default branch was among the ones just
    ///   pushed, `HEAD` resolves and this does nothing;
    /// * it considers only `refs/heads/…` that the push **created or moved**,
    ///   never a delete and never a tag; and
    /// * the choice goes through the catalog and then through
    ///   [`Self::point_head`], so the authority and its projection still move
    ///   together and there is still one writer of `HEAD`.
    ///
    /// It is logged at `warn`, not `info`: the server chose a default branch
    /// nobody asked for, and the operator has to be able to find that decision
    /// later. `set-default-branch` overrides it.
    pub fn adopt_default_branch_if_unborn(
        &self,
        path: &RepoRef,
        pushed: &[String],
    ) -> Result<Option<String>> {
        let Some(repo) = self.catalog().get(path)? else {
            // The repository was deleted between the push and here. Not an
            // error, and not something to invent a default branch for.
            return Ok(None);
        };
        let store = self.repo_store(&repo)?;
        // `HEAD` is not a row in `refs()` (the `GitServe` HEAD ruling), so this is the accessor
        // pair plus one resolution against the same snapshot the resolution
        // reads — see [`gunnar_wire::RefSnapshot::head`].
        // `FullName::new(HEAD_REF)` died with `RefStore`; there is nothing to
        // validate here, because the name is not a caller's.
        let git = store.serve();
        if gunnar_wire::RefSnapshot::read(git)?.head(git)?.is_some() {
            return Ok(None);
        }
        let Some(adopted) = pushed
            .iter()
            .find(|name| name.starts_with("refs/heads/"))
            .cloned()
        else {
            return Ok(None);
        };
        let repo = self.catalog().set_default_branch(path, &adopted)?;
        Self::point_head(&store, &repo)?;
        tracing::warn!(
            repo = %repo.key(),
            default_branch = %adopted,
            "this repository's default branch did not exist, so nothing could clone it; the \
             first branch pushed has been adopted as the default. `gunnar control repo \
             set-default-branch` overrides this"
        );
        Ok(Some(adopted))
    }

    /// **`repo create` is transactional, and it never adopts a store it did
    /// not make.** The one verb every create routes through: the gRPC
    /// `Repos.Create`, the local `gunnar repo create`, and the
    /// first-repository door once a push has passed the hash check.
    ///
    /// Order, and why each step is where it is:
    ///
    /// 1. **Look at the disk first.** If a directory already stands at the
    ///    name — or a [tombstone](Self::tombstone_path) says a delete could not
    ///    reclaim it — the create is refused **by name**, before the catalog is
    ///    touched. This is the 2026-08-30 resurrection: on v1.1.3-166 `repo
    ///    delete` dropped the catalog row and left the store, and the next
    ///    `repo create` of the same name opened it — `ls-remote` right after
    ///    the create listed all 59 refs of the deleted skade, and recreated
    ///    `--hash sha256` the name served sha1 refs (`fatal: protocol
    ///    error`). `adopt` is the opt-in for the one legitimate case (a
    ///    catalog restored from backup over a surviving data disk), and only
    ///    for a store whose [hash marker](Self::HASH_MARKER) matches.
    /// 2. The catalog row.
    /// 3. The store. If the engine cannot make it, step 2 is **rolled back**
    ///    and whatever the engine left is removed: a create that failed
    ///    leaves no directory, no row, nothing for `repo get` to find.
    ///
    /// Returns the catalogued repository; the handle is in the cache.
    pub fn create_repo(&self, spec: NewRepo, adopt: bool) -> Result<Repo> {
        let path = spec.path.clone();
        // TRUE when the directory was already there and was taken over. The
        // rollback below must not remove such a directory: "a create that
        // failed leaves no directory" is a promise about a directory THIS
        // create made, and applying it to an adopted store would answer a
        // transient engine failure by destroying the very bytes the operator
        // ran `--adopt` to keep. On the appliance that is 25-39 GB.
        let adopted = self.refuse_stale_store(&path, adopt, spec.hash_kind)?;
        let repo = self.catalog().create(spec)?;
        match self.repo_store(&repo) {
            Ok(_) => Ok(repo),
            Err(err) => {
                // Roll back: the row, the handle, the directory. A failure
                // inside the rollback is logged and the original error is the
                // one the caller sees, because that is the one they can act on.
                self.close_repo(&path);
                if let Err(undo) = self.catalog().delete(&path) {
                    tracing::error!(
                        target: "gunnar::store",
                        repo = %path,
                        error = format!("{undo:#}"),
                        "rolling back the catalog row of a create whose store failed"
                    );
                }
                if adopted {
                    tracing::error!(
                        target: "gunnar::store",
                        event = "repo.adopt_failed",
                        repo = %path,
                        "the create that ADOPTED this store failed to open it; the catalog row is \
                         rolled back and the DIRECTORY IS LEFT WHERE IT IS — it was not made by \
                         this create and is not this create's to remove"
                    );
                } else if let Ok(dir) = self.repo_dir(&path) {
                    if let Err(undo) = std::fs::remove_dir_all(&dir) {
                        if undo.kind() != std::io::ErrorKind::NotFound {
                            tracing::error!(
                                target: "gunnar::store",
                                repo = %path,
                                dir = %dir.display(),
                                error = %undo,
                                "rolling back the directory of a create whose store failed"
                            );
                        }
                    }
                }
                Err(err.context(if adopted {
                    format!(
                        "creating the stores of {path} over an ADOPTED directory; the row was \
                         rolled back and the store's bytes were left untouched"
                    )
                } else {
                    format!("creating the stores of {path}; nothing was kept")
                }))
            }
        }
    }

    /// The file that says which hash a store's objects are: `sha1` or
    /// `sha256`, one line, written when the store is first opened. Not the
    /// engine's — a store carries no marker of its own, and opening a sha1
    /// archive as sha256 does not fail, it decodes garbage. `--adopt` reads
    /// it; nothing else does.
    pub const HASH_MARKER: &'static str = "OBJECT_FORMAT";

    /// Step 1 of [`Self::create_repo`]: nothing may stand at `path` on disk.
    ///
    /// Returns **true when an existing store was ADOPTED** — the one case in
    /// which the directory was not made by this create, and therefore the one
    /// case in which [`Self::create_repo`]'s rollback must not remove it.
    ///
    /// ## `--adopt` means "take it as it is", including a store from before
    /// the marker existed
    ///
    /// MEASURED on the appliance 2026-08-30, `v1.1.3-227-g6a2367a`, request
    /// `175daccc0000000d`:
    ///
    /// ```text
    /// repo create alice/linux --hash sha1 --public --adopt
    ///   -> failed precondition: cannot adopt the store at
    ///      /data/repos/alice/linux.git: it carries unmarked objects, the
    ///      create asked for sha1
    /// ```
    ///
    /// The marker is a 2026-08-30 file. **Every store written before that day
    /// has none**, so an `--adopt` that requires one refuses precisely the
    /// stores adoption exists for — a catalog restored over a surviving data
    /// disk, or the orphan an old catalog-only delete left. Composed with a
    /// `create` that refuses the name and a `delete` that cannot authorise it,
    /// that is a store with no way out at all.
    ///
    /// The honest rule, and the one implemented here:
    ///
    /// * **unmarked** — there is no second opinion to check, so the caller's
    ///   `--hash` is taken at their word AND WRITTEN into the marker. The
    ///   store is marked from then on, so the *next* adopt is checked; the
    ///   caller's word is recorded rather than merely trusted.
    /// * **marked and equal** — adopted, as before.
    /// * **marked and different** — refused, as before. This is the arm that
    ///   stops a sha256 row over a sha1 store from serving refs no client can
    ///   read, and it is not weakened: a store that carries a marker is
    ///   checked against it.
    pub fn refuse_stale_store(
        &self,
        path: &RepoRef,
        adopt: bool,
        hash_kind: HashKind,
    ) -> Result<bool> {
        let dir = self.repo_dir(path)?;
        let tombstone = Self::tombstone_path(&dir);
        let tombstoned = tombstone.exists();
        if !dir.exists() && !tombstoned {
            return Ok(false);
        }
        if !adopt {
            let why = if tombstoned {
                let record =
                    Tombstone::parse(&std::fs::read_to_string(&tombstone).unwrap_or_default());
                match record.reason {
                    // **The soft-delete window.** The name is unusable because
                    // the repository is still there, being kept — and the
                    // sentence has to say `undelete`, because a create is
                    // exactly what somebody reaches for when they mean it.
                    TombstoneReason::Deleted => {
                        let left = record
                            .seconds_left(gunnar_core::now_ms())
                            .map(|s| format!("{} day(s) left", s / 86_400))
                            .unwrap_or_else(|| "no expiry set".to_owned());
                        format!(
                            "{path} is DELETED but not purged: its bytes are being kept ({left}). \
                             `repo undelete {path}` puts it back as it was, `repo purge {path}` \
                             reclaims it now, or pass --adopt to take the store as a new \
                             repository and lose the old row"
                        )
                    }
                    TombstoneReason::ReclaimFailed => format!(
                        "a delete of this name could not reclaim its store ({}); `repo delete \
                         {path} --purge` retries the reclaim, or pass --adopt to take the store \
                         as it is",
                        record.note.trim()
                    ),
                }
            } else {
                format!(
                    "a store already exists at {}; `repo delete {path}` reclaims it, or pass \
                     --adopt to take it as it is",
                    dir.display()
                )
            };
            anyhow::bail!("{why}");
        }
        let marker = dir.join(Self::HASH_MARKER);
        let on_disk = std::fs::read_to_string(&marker)
            .map(|text| text.trim().to_owned())
            .unwrap_or_default();
        if on_disk.is_empty() {
            // Unmarked: a store from before the marker existed. There is
            // nothing to disagree with, so the caller's word is taken — and
            // WRITTEN, so it is a fact from here on rather than a decision
            // taken again on every adopt.
            std::fs::write(&marker, format!("{hash_kind}\n")).with_context(|| {
                format!(
                    "marking the adopted store at {} as {hash_kind}; the adopt is refused rather \
                     than leaving a store whose hash nothing records",
                    dir.display()
                )
            })?;
            tracing::warn!(
                target: "gunnar::store",
                event = "repo.adopted_unmarked",
                repo = %path,
                dir = %dir.display(),
                hash = %hash_kind,
                marker = %marker.display(),
                "an UNMARKED store (written before the OBJECT_FORMAT marker existed) was adopted \
                 at the caller's word; it is marked {} from now on and the next adopt is CHECKED",
                hash_kind
            );
        } else {
            anyhow::ensure!(
                on_disk == hash_kind.to_string(),
                "cannot adopt the store at {}: its objects are {} (marker {} says so), the \
                 create asked for {hash_kind}. Adopt it as {} or leave it alone; a {hash_kind} \
                 row over a {} store serves refs no client can read.",
                dir.display(),
                on_disk,
                marker.display(),
                on_disk,
                on_disk
            );
        }
        if tombstoned {
            // Framed, as every tombstone change is (`frame_tombstone`): an adopt over a
            // soft-deleted name (and `repo undelete`, which adopts) must not leave a box
            // restored from the twin with the name still fenced off.
            let _framed = self.frame_tombstone(&tombstone, None)?;
            std::fs::remove_file(&tombstone)
                .with_context(|| format!("removing the tombstone {}", tombstone.display()))?;
        }
        tracing::warn!(
            target: "gunnar::store",
            event = "repo.adopted",
            repo = %path,
            dir = %dir.display(),
            hash = %hash_kind,
            "an existing store was ADOPTED by a create with --adopt; its objects are served as they are"
        );
        Ok(true)
    }

    /// Where a name's tombstone lives: beside its directory, so a store that
    /// could not be removed and its marker are removed together by the retry.
    pub fn tombstone_path(dir: &Path) -> PathBuf {
        let mut name = dir
            .file_name()
            .map(|n| n.to_os_string())
            .unwrap_or_default();
        name.push(".tombstone");
        dir.with_file_name(name)
    }

    /// Is `path` tombstoned — deleted from the catalog, store not reclaimed?
    pub fn is_tombstoned(&self, path: &RepoRef) -> Result<bool> {
        Ok(Self::tombstone_path(&self.repo_dir(path)?).exists())
    }

    /// How many repository handles are open. For tests and `Admin.Stats`.
    pub fn open_handles(&self) -> usize {
        self.cache().entries.len()
    }

    /// Forget a repository's handle, releasing its store lock. Used by
    /// `repo delete --purge` before the directory is removed.
    pub fn close_repo(&self, path: &RepoRef) {
        self.usage.mark_repo(path.namespace(), path.name());
        self.cache().entries.remove(&path.key());
    }

    /// **`repo delete` reclaims the disk.** Stop the repository's index work,
    /// release its handle, remove its directory — pack, journal, tail,
    /// exploded tier and index tables — and report the bytes that went.
    ///
    /// MEASURED 2026-08-29 on the appliance: a `repo delete` after the linux
    /// push left **25 GB** of `gunnar-git.znippy/` behind, because the catalog
    /// row was removed and the directory was not. Order here is the point:
    ///
    /// 1. the engine cancels the index worker (an in-flight pack is **aborted**
    ///    at its next checkpoint, milliseconds, never finished — its rows are
    ///    keyed by oid and the directory is about to go);
    /// 2. the handle cache forgets the repository, so the store drops as soon as
    ///    the last request holding it returns (the worker is cancelled, so the
    ///    drop does not join hours of work);
    /// 3. `remove_dir_all`. A store still held by an in-flight request keeps
    ///    its open fds on unlinked files until it returns; the bytes are
    ///    reclaimed when it does, and the directory entry is gone now.
    ///
    /// Returns `(bytes_before, bytes_after)` measured by walking the directory,
    /// so the log line and the caller report applied output.
    ///
    /// **A reclaim that fails leaves the name TOMBSTONED** (2026-08-30): a
    /// file beside the directory saying when and why, so
    /// [`Self::create_repo`] refuses the name until `repo delete` is run again
    /// and succeeds. Without it a failed `remove_dir_all` (a store held open
    /// on a read-only mount, a permissions slip) left a directory the next
    /// create would refuse by name anyway — but with no sentence saying that
    /// a delete had been asked for and had not finished.
    pub fn purge_repo(&self, path: &RepoRef) -> Result<(u64, u64)> {
        // A snapshot streaming from its cut may still read this store (R5.1-A).
        self.wait_unpinned("removing a repository store");
        let _usage = UsageMark::namespace(&self.usage, path.namespace());
        // **A `RepoStore` purge frame before the directory goes** (twin note, M9): the twin
        // removes the same store at the same point of the sequence, and a restore replays the
        // removal, so a purged repository is never resurrected from the log's earlier packs.
        // A twin applying the frame reaches `reclaim_repo` directly and frames nothing.
        if !self.txlog.is_follower() {
            let dir = self.repo_dir(path)?;
            let hash = std::fs::read_to_string(dir.join(Self::HASH_MARKER))
                .map(|s| s.trim().to_string())
                .unwrap_or_default();
            let meta = gunnar_core::txlog::Meta::RepoStore {
                repo: path.key(),
                op: "purged".into(),
                hash,
            };
            let _framed = self.txlog.mutation();
            self.txlog
                .append(gunnar_core::txlog::Kind::RepoStore, &meta, &[])
                .with_context(|| format!("framing the purge of {path}"))?;
            return self.reclaim_repo(path);
        }
        self.reclaim_repo(path)
    }

    /// The removal half of [`Self::purge_repo`], unframed: what the twin's apply of a
    /// `RepoStore` purge frame runs.
    pub fn reclaim_repo(&self, path: &RepoRef) -> Result<(u64, u64)> {
        let _usage = UsageMark::namespace(&self.usage, path.namespace());
        let dir = self.repo_dir(path)?;
        let tombstone = Self::tombstone_path(&dir);
        let before = dir_bytes(&dir);
        self.engine.cancel_index(&dir);
        self.close_repo(path);
        match std::fs::remove_dir_all(&dir) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                let now = gunnar_core::now_ms();
                let record = Tombstone {
                    deleted_at_unix_ms: now,
                    // No deadline: a reclaim that failed is OWED to an
                    // operator, and a sweep that retried it unattended would
                    // be repeating the thing that just did not work.
                    purge_after_unix_ms: None,
                    principal_id: String::new(),
                    reason: TombstoneReason::ReclaimFailed,
                    repo: None,
                    note: format!("delete asked at unix_ms {now}, reclaim failed: {e}"),
                };
                let note = record.note.clone();
                let text = serde_json::to_string_pretty(&record)
                    .unwrap_or_else(|_| format!("{{\"note\":\"{note}\"}}"));
                if let Err(write) = std::fs::write(&tombstone, format!("{text}\n")) {
                    tracing::error!(
                        target: "gunnar::store",
                        repo = %path,
                        tombstone = %tombstone.display(),
                        error = %write,
                        "the reclaim failed AND the tombstone could not be written"
                    );
                }
                tracing::error!(
                    target: "gunnar::store",
                    event = "repo.purge_failed",
                    repo = %path,
                    dir = %dir.display(),
                    error = %e,
                    "repo delete could not reclaim the store; the name is tombstoned and a create \
                     refuses it until `repo delete` is run again and succeeds"
                );
                return Err(anyhow::anyhow!(
                    "the catalog row is gone but the store at {} could not be reclaimed ({e}); \
                     the name is tombstoned — `repo delete {path}` retries, `repo create` refuses \
                     it until then",
                    dir.display()
                ));
            }
        }
        if tombstone.exists() {
            std::fs::remove_file(&tombstone)
                .with_context(|| format!("removing the tombstone {}", tombstone.display()))?;
        }
        let after = dir_bytes(&dir);
        tracing::info!(
            target: "gunnar::store",
            event = "repo.purged",
            repo = %path,
            dir = %dir.display(),
            bytes_before = before,
            bytes_after = after,
            "repo delete reclaimed {} bytes",
            before.saturating_sub(after)
        );
        Ok((before, after))
    }

    // ── the soft-delete window (2026-08-30) ────────────────────────────────

    /// **`repo delete` without `--purge`: a DATED TOMBSTONE, not a reclaim.**
    ///
    /// The catalog row is the caller's to remove; this is everything else. The
    /// bytes stay exactly where they are, the name becomes unusable
    /// ([`Self::refuse_stale_store`] refuses a create over a tombstone), and a
    /// deadline is written into the tombstone at this instant.
    ///
    /// Why this exists at all, in one measured line from
    /// MEASURED: `repo delete --purge` returns the bytes in
    /// ~2 seconds — **181 GiB to 33 MiB** — and a live mirror does not help
    /// because the delete replicates. The failure a first paying customer is
    /// most likely to cause is deleting their own repository, and the only
    /// thing that answers it is a window.
    ///
    /// What it does NOT do, and deliberately:
    ///
    /// * it does not touch the bytes, so it cannot fail on a full disk;
    /// * it cancels the index worker and drops the handle, because a
    ///   repository nobody may reach should not be spending CPU or file
    ///   descriptors on being resolved;
    /// * it does not delete the tombstone of a *failed reclaim* — that record
    ///   is owed to an operator. Soft-deleting a name already tombstoned that
    ///   way is refused by name.
    ///
    /// `retention` of zero is legal and means *purge at the next sweep*: an
    /// operator reclaiming space said so, and a zero that silently became
    /// thirty days would be the server overruling them.
    pub fn soft_delete(
        &self,
        path: &RepoRef,
        repo: Option<&Repo>,
        principal_id: &str,
        retention: std::time::Duration,
    ) -> Result<Tombstone> {
        let _usage = UsageMark::namespace(&self.usage, path.namespace());
        let dir = self.repo_dir(path)?;
        let marker = Self::tombstone_path(&dir);
        if let Some(existing) = Self::read_tombstone(&marker) {
            if existing.reason == TombstoneReason::ReclaimFailed {
                anyhow::bail!(
                    "{path} is already tombstoned because a reclaim FAILED ({}). That is bytes \
                     owed, not bytes held: run `repo delete {path} --purge` to finish it. \
                     Soft-deleting it again would hide the failure behind a deadline.",
                    existing.note.trim()
                );
            }
        }
        anyhow::ensure!(
            dir.is_dir(),
            "{path} has nothing on disk at {}; there is nothing to keep, and a tombstone over \
             no bytes would promise a restore that cannot happen",
            dir.display()
        );
        let now = gunnar_core::now_ms();
        let record = Tombstone {
            deleted_at_unix_ms: now,
            purge_after_unix_ms: Some(now + retention.as_millis() as i64),
            principal_id: principal_id.to_owned(),
            reason: TombstoneReason::Deleted,
            repo: repo.cloned(),
            note: format!(
                "soft-deleted at unix_ms {now}; the bytes are kept until the deadline, `repo \
                 undelete {path}` restores it"
            ),
        };
        // The engine first: a repository nobody may reach must not keep
        // resolving a pack. The handle after it, so the store drops as soon as
        // the last in-flight request returns.
        self.engine.cancel_index(&dir);
        self.close_repo(path);
        let text = serde_json::to_string_pretty(&record).context("serialising the tombstone")?;
        let text = format!("{text}\n");
        // **The twin hears of it FIRST** (GHUNT-SOFTDEL, gunnar.rs 2026-10-28): a `File` frame
        // before the file, as `durable::write_atomic` does for every document under the data
        // root. Unframed, the twin held the catalog row's removal and not the tombstone, so a
        // box restored from it (the recharge, a main rebuilt from its twin) held the bytes as
        // an ORPHAN: not listed by `repo list-deleted`, and no `repo undelete`.
        let _framed = self.frame_tombstone(&marker, Some(text.as_bytes()))?;
        std::fs::write(&marker, &text)
            .with_context(|| format!("writing the tombstone {}", marker.display()))?;
        let bytes = dir_bytes(&dir);
        tracing::warn!(
            target: "gunnar::store",
            event = "repo.soft_deleted",
            repo = %path,
            dir = %dir.display(),
            bytes,
            principal = %principal_id,
            purge_after_unix_ms = record.purge_after_unix_ms.unwrap_or_default(),
            "repo delete tombstoned this repository; {bytes} bytes are KEPT until the retention \
             window ends, and they still count against the account"
        );
        Ok(record)
    }

    /// **`repo undelete`: put a soft-deleted repository back.**
    ///
    /// Returns the [`NewRepo`] the caller must re-insert into the catalog, and
    /// removes the tombstone. The two halves are split because the catalog is
    /// the caller's — the gRPC path and the local CLI reach it differently —
    /// and because the row must go in *before* the tombstone comes off: a
    /// crash between them leaves a repository that is still fenced off, which
    /// is recoverable, rather than a name a create would adopt, which is the
    /// resurrection the reclaim rules closed.
    ///
    /// The recorded hash kind is the one that comes back, never a default. A
    /// tombstone that does not carry a row (one written before this existed)
    /// is refused BY NAME rather than restored as sha256 over a sha1 store.
    pub fn undelete_spec(&self, path: &RepoRef) -> Result<NewRepo> {
        let dir = self.repo_dir(path)?;
        let marker = Self::tombstone_path(&dir);
        let record = Self::read_tombstone(&marker).ok_or_else(|| {
            anyhow::anyhow!(
                "{path} is not deleted: there is no tombstone at {}. `repo list-deleted` shows \
                 what can be undeleted.",
                marker.display()
            )
        })?;
        anyhow::ensure!(
            record.reason == TombstoneReason::Deleted,
            "{path} is tombstoned because a RECLAIM FAILED, not because it was deleted with a \
             window ({}). There is nothing to undelete: the operator asked for these bytes to \
             go. `repo delete {path} --purge` finishes it.",
            record.note.trim()
        );
        anyhow::ensure!(
            dir.is_dir(),
            "{path} is tombstoned but its bytes are gone from {}; there is nothing to restore",
            dir.display()
        );
        let repo = record.repo.ok_or_else(|| {
            anyhow::anyhow!(
                "the tombstone at {} does not record what {path} was — no hash kind, no default \
                 branch. Restoring it would be a GUESS, and a sha256 row over a sha1 store \
                 serves refs no client can read. Recreate it with `repo create --adopt --hash \
                 <the store's OBJECT_FORMAT>` instead.",
                marker.display()
            )
        })?;
        Ok(NewRepo {
            path: path.clone(),
            hash_kind: repo.hash_kind,
            default_branch: Some(repo.default_branch),
            visibility: repo.visibility,
            description: repo.description,
        })
    }

    /// Take the tombstone off, once the catalog row is back.
    pub fn clear_tombstone(&self, path: &RepoRef) -> Result<()> {
        let _usage = UsageMark::namespace(&self.usage, path.namespace());
        let marker = Self::tombstone_path(&self.repo_dir(path)?);
        // Framed before it goes, for the reason `soft_delete` frames its write: a box restored
        // from the twin must not find the undeleted repository still fenced off.
        let _framed = if marker.exists() {
            self.frame_tombstone(&marker, None)?
        } else {
            None
        };
        match std::fs::remove_file(&marker) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(e)
                    .with_context(|| format!("removing the tombstone {}", marker.display()))
            }
        }
        tracing::warn!(
            target: "gunnar::store",
            event = "repo.undeleted",
            repo = %path,
            "repo undelete put this repository back; its bytes were never reclaimed"
        );
        Ok(())
    }

    /// **A tombstone's write (`Some(bytes)`) or removal (`None`) as a `File` frame** in this
    /// box's log, before the file changes; the guard holds the log's mutation lock until the
    /// caller has changed it. Nothing on a follower (a twin or a box being restored: it applies
    /// frames, it originates none).
    fn frame_tombstone(
        &self,
        marker: &Path,
        bytes: Option<&[u8]>,
    ) -> Result<Option<gunnar_core::txlog::MutationGuard>> {
        if self.txlog.is_follower() {
            return Ok(None);
        }
        let rel = marker
            .strip_prefix(&self.root)
            .with_context(|| {
                format!(
                    "the tombstone {} is not under the data root",
                    marker.display()
                )
            })?
            .to_string_lossy()
            .into_owned();
        let guard = self.txlog.mutation();
        self.txlog
            .append(
                gunnar_core::txlog::Kind::File,
                &gunnar_core::txlog::Meta::File {
                    path: rel.clone(),
                    deleted: bytes.is_none(),
                },
                bytes.unwrap_or_default(),
            )
            .with_context(|| format!("framing the tombstone {rel}"))?;
        Ok(Some(guard))
    }

    /// Read a tombstone file, or `None` when there is not one.
    pub fn read_tombstone(marker: &Path) -> Option<Tombstone> {
        std::fs::read_to_string(marker)
            .ok()
            .map(|text| Tombstone::parse(&text))
    }

    /// The tombstone beside `path`, if any.
    pub fn tombstone_of(&self, path: &RepoRef) -> Result<Option<Tombstone>> {
        Ok(Self::read_tombstone(&Self::tombstone_path(
            &self.repo_dir(path)?,
        )))
    }

    /// **Every soft-deleted repository, with the bytes it is still holding.**
    ///
    /// This is what makes a tombstone visible on a bill: the bytes are real,
    /// the account is still using them, and a customer who deleted a
    /// repository and saw no change in their usage would rightly call that a
    /// bug. `Admin.StorageInfo` reports the total and `repo list-deleted`
    /// prints the rows.
    ///
    /// It walks `repos/` rather than asking the catalog, because the whole
    /// point of a tombstone is that the catalog does not know about it.
    pub fn tombstones(&self) -> Result<Vec<DeletedRepo>> {
        let root = self.repos_root().to_path_buf();
        let mut found = Vec::new();
        Self::walk_tombstones(&root, &root, &mut found);
        found.sort_by_key(|a| a.path.key());
        Ok(found)
    }

    /// The recursion behind [`Self::tombstones`]. Namespaces nest, so
    /// this cannot be one `read_dir`.
    fn walk_tombstones(root: &Path, dir: &Path, out: &mut Vec<DeletedRepo>) {
        let Ok(entries) = std::fs::read_dir(dir) else {
            return;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            let name = entry.file_name().to_string_lossy().into_owned();
            if name.ends_with(".git.tombstone") {
                let Some(record) = Self::read_tombstone(&path) else {
                    continue;
                };
                if record.reason != TombstoneReason::Deleted {
                    continue;
                }
                let store = path.with_file_name(name.trim_end_matches(".tombstone"));
                let Some(repo_ref) = Self::repo_ref_of_dir(root, &store) else {
                    continue;
                };
                out.push(DeletedRepo {
                    path: repo_ref,
                    tombstone: record,
                    bytes: dir_bytes(&store),
                });
                continue;
            }
            if path.is_dir() && !name.ends_with(".git") {
                Self::walk_tombstones(root, &path, out);
            }
        }
    }

    /// `<repos-root>/<ns…>/<name>.git` back to a [`RepoRef`].
    fn repo_ref_of_dir(root: &Path, dir: &Path) -> Option<RepoRef> {
        let rel = dir.strip_prefix(root).ok()?;
        let mut parts: Vec<String> = rel
            .components()
            .map(|c| c.as_os_str().to_string_lossy().into_owned())
            .collect();
        let name = parts.pop()?;
        let name = name.strip_suffix(".git").unwrap_or(&name).to_owned();
        if parts.is_empty() {
            return None;
        }
        RepoRef::new(&parts.join("/"), &name).ok()
    }

    // ── orphan stores: the directory nothing can see (2026-08-30) ──────────

    /// **Every store directory with no catalog row and no soft-delete
    /// tombstone.**
    ///
    /// MEASURED on the appliance 2026-08-30 (`v1.1.3-227-g6a2367a`), the
    /// wreckage of Friday's kernel-mirror incident: `/data/repos/alice/linux.git`
    /// stood on the disk with ~25-39 GB in it, no catalog row and no
    /// tombstone. Three guards, each right on its own, composed into a state
    /// with no exit — `create` refused the name because the directory was
    /// there (`175daccc0000000c`), `create --adopt` refused it because the
    /// store predated the `OBJECT_FORMAT` marker (`…0d`), and `delete
    /// --purge` refused it because the access check was `Access::repo` on a
    /// repository not in the catalog, which no principal can ever hold a grant
    /// on (`…0b`/`…0e`). The bytes were stranded and the NAME was permanently
    /// unusable.
    ///
    /// This is the door that was missing. It walks `repos/` rather than
    /// asking the catalog, exactly as [`Self::tombstones`] does and for the
    /// same reason: the whole point of an orphan is that no record of it
    /// exists.
    ///
    /// What counts as an orphan, and why it is drawn here:
    ///
    /// * **no catalog row** — a catalogued directory is `repo list`'s;
    /// * **no `Deleted` tombstone** — one of those is `repo list-deleted`'s,
    ///   and it has `undelete` and `purge` as its exits;
    /// * a `ReclaimFailed` tombstone IS an orphan. `tombstones()` filters that
    ///   reason out of `list-deleted` deliberately (bytes owed, not bytes
    ///   held), which left it in no listing whatsoever; the row carries the
    ///   note so an operator sees why.
    pub fn orphans(&self) -> Result<Vec<OrphanStore>> {
        let root = self.repos_root().to_path_buf();
        let mut found = Vec::new();
        self.walk_orphans(&root, &root, &mut found)?;
        found.sort_by_key(|a| a.path.key());
        Ok(found)
    }

    /// The recursion behind [`Self::orphans`]. Namespaces nest, so this
    /// cannot be one `read_dir` — the same shape as
    /// [`Self::walk_tombstones`].
    fn walk_orphans(&self, root: &Path, dir: &Path, out: &mut Vec<OrphanStore>) -> Result<()> {
        let Ok(entries) = std::fs::read_dir(dir) else {
            return Ok(());
        };
        for entry in entries.flatten() {
            let path = entry.path();
            let name = entry.file_name().to_string_lossy().into_owned();
            if !path.is_dir() {
                continue;
            }
            if !name.ends_with(".git") {
                self.walk_orphans(root, &path, out)?;
                continue;
            }
            let Some(repo_ref) = Self::repo_ref_of_dir(root, &path) else {
                continue;
            };
            if self.catalog().get(&repo_ref)?.is_some() {
                continue;
            }
            let record = Self::read_tombstone(&Self::tombstone_path(&path));
            let failed_reclaim = match &record {
                // `repo list-deleted` owns this one; it is not an orphan.
                Some(t) if t.reason == TombstoneReason::Deleted => continue,
                Some(t) => Some(t.note.trim().to_owned()),
                None => None,
            };
            let hash_marker = std::fs::read_to_string(path.join(Self::HASH_MARKER))
                .ok()
                .map(|text| text.trim().to_owned())
                .filter(|text| !text.is_empty());
            out.push(OrphanStore {
                path: repo_ref,
                bytes: dir_bytes(&path),
                modified_unix_ms: newest_mtime_unix_ms(&path),
                hash_marker,
                failed_reclaim,
                dir: path,
            });
        }
        Ok(())
    }

    /// **Is there a store at `path` that no record accounts for?**
    ///
    /// The one question both delete doors ask, so they cannot answer it
    /// differently (one implementation): a directory on disk, no catalog row, and no
    /// `Deleted` tombstone. A `ReclaimFailed` tombstone still counts as an
    /// orphan — those bytes are owed to an operator and no listing but
    /// [`Self::orphans`] shows them.
    pub fn is_orphan(&self, path: &RepoRef) -> Result<bool> {
        if self.catalog().get(path)?.is_some() {
            return Ok(false);
        }
        let dir = self.repo_dir(path)?;
        if !dir.is_dir() {
            return Ok(false);
        }
        Ok(!matches!(
            Self::read_tombstone(&Self::tombstone_path(&dir)),
            Some(record) if record.reason == TombstoneReason::Deleted
        ))
    }

    /// **What a delete of an orphan may and may not do**, as one sentence both
    /// doors print.
    ///
    /// `--purge` on an orphan is legitimate and is exactly what `repo reclaim`
    /// does: the operator said "the bytes go", the row is already gone, the
    /// bytes are there. A *soft* delete of one is not, and the reason is the
    /// soft delete's own contract — it promises `repo undelete` puts the
    /// repository back **as it was**, and for an orphan the server does not
    /// know what it was: no hash kind, no default branch, no visibility. Lane
    /// AC already refuses to restore a tombstone that carries no row; refusing
    /// to WRITE one is that same rule, one step earlier, before a deadline has
    /// been put on bytes nobody can bring back.
    pub fn orphan_soft_delete_refusal(path: &RepoRef) -> String {
        format!(
            "{path} is an ORPHAN store: bytes on disk with no catalog row and no tombstone, so \
             the server does not know what it was and a tombstone would promise an undelete it \
             cannot honour. Three exits, and they are all one command: `repo delete {path} \
             --purge` reclaims the bytes now, `repo reclaim {path}` is the same thing said \
             plainly, and `repo create {path} --adopt --hash <its OBJECT_FORMAT>` takes the \
             store over as a repository instead. `repo list-orphans` reports that hash."
        )
    }

    /// **Reclaim an orphan's bytes.** The exit that did not exist.
    ///
    /// It refuses anything that is not an orphan BY NAME, because both of the
    /// other two states have their own verb and reaching them through this one
    /// would be a second way to destroy a repository:
    ///
    /// * catalogued -> `repo delete --purge`;
    /// * soft-deleted -> `repo undelete` or `repo purge`.
    ///
    /// The reclaim itself is [`Self::purge_repo`] — the same verb `--purge`
    /// and the sweep use, so an orphan's bytes go the way every other
    /// repository's bytes go (one implementation), including the index worker being
    /// cancelled first and a `ReclaimFailed` tombstone if it cannot.
    pub fn reclaim_orphan(&self, path: &RepoRef) -> Result<u64> {
        let _usage = UsageMark::namespace(&self.usage, path.namespace());
        anyhow::ensure!(
            self.catalog().get(path)?.is_none(),
            "{path} is a LIVE repository, not an orphan: it is in the catalog. `repo delete \
             {path} --purge` is the verb that deletes it."
        );
        let dir = self.repo_dir(path)?;
        if let Some(record) = Self::read_tombstone(&Self::tombstone_path(&dir)) {
            anyhow::ensure!(
                record.reason != TombstoneReason::Deleted,
                "{path} is SOFT-DELETED, not an orphan: its bytes are being kept on purpose and \
                 `repo undelete {path}` still puts it back. `repo purge {path}` ends the window."
            );
        }
        anyhow::ensure!(
            dir.is_dir(),
            "there is no store at {}; `repo list-orphans` shows what there is to reclaim",
            dir.display()
        );
        let (before, after) = self.purge_repo(path)?;
        let bytes = before.saturating_sub(after);
        tracing::warn!(
            target: "gunnar::store",
            event = "repo.orphan_reclaimed",
            repo = %path,
            dir = %dir.display(),
            bytes,
            "an ORPHAN store — no catalog row, no tombstone — was reclaimed; {bytes} bytes back \
             and the name is usable again"
        );
        Ok(bytes)
    }

    /// **The sweep: reclaim every tombstone whose window has closed.**
    ///
    /// Returns what it reclaimed, `(path, bytes)`. A failure on one repository
    /// is logged and the sweep continues — one undeletable directory must not
    /// stop the other nineteen from being reclaimed, and `purge_repo` leaves
    /// its own `ReclaimFailed` tombstone behind so the failure is not lost.
    pub fn sweep_expired_tombstones(&self, now_unix_ms: i64) -> Vec<(RepoRef, u64)> {
        let deleted = match self.tombstones() {
            Ok(deleted) => deleted,
            Err(err) => {
                tracing::error!(
                    target: "gunnar::store",
                    error = format!("{err:#}"),
                    "the tombstone sweep could not list what is deleted; nothing was reclaimed"
                );
                return Vec::new();
            }
        };
        let mut reclaimed = Vec::new();
        for row in deleted {
            if !row.tombstone.is_expired_at(now_unix_ms) {
                continue;
            }
            match self.purge_repo(&row.path) {
                Ok((before, after)) => {
                    let bytes = before.saturating_sub(after);
                    tracing::warn!(
                        target: "gunnar::store",
                        event = "repo.tombstone_swept",
                        repo = %row.path,
                        bytes,
                        deleted_at_unix_ms = row.tombstone.deleted_at_unix_ms,
                        "the retention window closed; a soft-deleted repository was reclaimed \
                         and cannot be undeleted any more"
                    );
                    reclaimed.push((row.path, bytes));
                }
                Err(err) => {
                    tracing::error!(
                        target: "gunnar::store",
                        event = "repo.tombstone_sweep_failed",
                        repo = %row.path,
                        error = format!("{err:#}"),
                        "an expired tombstone could not be reclaimed; the name stays fenced off \
                         and the sweep continues"
                    );
                }
            }
        }
        reclaimed
    }

    /// Which catalogued repository the git directory `dir` belongs to, as
    /// `namespace/name` — the inverse of [`Self::repo_dir`], for a progress
    /// row that knows only a store root.
    pub fn repo_of_dir(&self, dir: &Path) -> Option<(String, String)> {
        let rel = dir.strip_prefix(&self.repos_root).ok()?;
        let mut parts = rel.components().filter_map(|c| match c {
            std::path::Component::Normal(s) => Some(s.to_string_lossy().into_owned()),
            _ => None,
        });
        let ns = parts.next()?;
        let name = parts.next()?;
        Some((ns, name.strip_suffix(".git").unwrap_or(&name).to_owned()))
    }
}

/// **Start the soft-delete sweep**: hourly, and once immediately.
///
/// The other half of a soft delete. Without it "deleted" would mean "kept
/// forever" and a `/data` would fill with repositories nobody can see — which
/// is the invisible 25 GB again, only with a promise attached.
///
/// It runs once at start on purpose: a box that was powered off when a
/// retention window closed must not wait an hour to notice, and an operator who
/// set `--retention-days 0` to free space should see it happen while they are
/// still watching.
///
/// Fire and forget, exactly like `disktest::spawn`. It holds no lock the server
/// needs, its work is on the blocking pool, and a failure on one repository is
/// logged and skipped rather than stopping the sweep — `purge_repo` leaves its
/// own tombstone behind, so nothing is lost by carrying on.
pub fn spawn_tombstone_sweep(stores: Arc<ServerStores>) -> tokio::task::JoinHandle<()> {
    // A32-14, stays a task: its handle is kept by serve and aborted at the server's stop.
    tokio::spawn(async move {
        loop {
            let stores = Arc::clone(&stores);
            let reclaimed = tokio::task::spawn_blocking(move || {
                stores.sweep_expired_tombstones(gunnar_core::now_ms())
            })
            .await
            .unwrap_or_default();
            if !reclaimed.is_empty() {
                let bytes: u64 = reclaimed.iter().map(|(_, b)| b).sum();
                tracing::warn!(
                    target: "gunnar::store",
                    event = "repo.sweep",
                    repos = reclaimed.len(),
                    bytes,
                    "the retention window closed on {} repository(ies); {bytes} bytes were \
                     reclaimed and they can no longer be undeleted",
                    reclaimed.len()
                );
            }
            tokio::time::sleep(TOMBSTONE_SWEEP_INTERVAL).await;
        }
    })
}

/// What one flow-flush tick did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FlowFlush {
    /// This many namespace-day rows written.
    Wrote(usize),
    /// Not attempted: the transaction log is RELEASED (a runtime data-set growth has given its
    /// files up and `/data` is unmounted). Nothing is lost — the live counters are in RAM and a
    /// refused write re-parks its completed days — and nothing is logged above trace.
    Held,
    /// The write failed for any other reason.
    Failed(String),
}

/// **One flow-flush tick.** Found by the lifecycle lane 2026-09-15: every minute of a runtime
/// growth logged `flow.flush_failed`, because the flush wrote through a transaction log that
/// had been released for the unmount. The release window is now a quiet hold, and a write
/// the release overtook mid-flush is read as the same hold rather than as a failure.
pub fn flow_flush_tick(stores: &ServerStores) -> FlowFlush {
    if stores.txlog().is_released() {
        return FlowFlush::Held;
    }
    match stores.flow().flush(gunnar_core::now_ms()) {
        Ok(rows) => FlowFlush::Wrote(rows),
        Err(_) if stores.txlog().is_released() => FlowFlush::Held,
        Err(err) => FlowFlush::Failed(err.to_string()),
    }
}

/// **Start the flow flush**: every [`FLOW_FLUSH_INTERVAL`], forever.
///
/// The live half of the flow ledger is a set of atomics in this process; this
/// is what makes it survive a restart. It COPIES the live slots into the
/// control store rather than adding to them (`gunnar_core::FlowLedger::flush`),
/// so running it twice writes the same rows twice instead of doubling them —
/// which is what lets it run on a tick without any coordination, and what lets
/// a slot opened after a restart be seeded from the stored row rather than from
/// zero.
///
/// It is a task of its own and NOT a line in the dashboard's two-second tick,
/// deliberately: that tick is documented as reading a few atomics and nothing
/// else, and this writes redb. A flush on the dashboard's thread would put a
/// database commit between the operator and the screen.
///
/// It also carries the day boundary on an IDLE server. A namespace that was
/// pushed to at 23:59 and then left alone has a live slot still holding
/// yesterday; nothing on the transfer path will notice, because there is no
/// transfer. The flush rolls it and writes the completed day.
///
/// Fire and forget, like [`spawn_tombstone_sweep`]: it holds no lock the server
/// needs, its work is on the blocking pool, and a failed write is logged and
/// retried on the next tick — the counters are still in RAM, so nothing is lost
/// by carrying on.
pub fn spawn_flow_flush(
    stores: Arc<ServerStores>,
    band: Arc<Mutex<Option<gunnar_core::Fleet>>>,
) -> tokio::task::JoinHandle<()> {
    // A32-14, stays a task: its handle is kept by serve and aborted at the server's stop.
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(FLOW_FLUSH_INTERVAL).await;
            let tick = Arc::clone(&stores);
            let band = Arc::clone(&band);
            let written = tokio::task::spawn_blocking(move || {
                let written = flow_flush_tick(&tick);
                if written == FlowFlush::Held {
                    // The fold reads the control document, which is under the unmounted
                    // data set too: the band keeps its last answer through the window.
                    return written;
                }
                // **The usage ledger's reconcile pass** (2026-10-06): one slice of
                // namespaces re-measured per tick, the whole set every
                // `RECONCILE_CYCLE`, so bytes that moved without a mark are
                // corrected and counted rather than billed stale. Low priority by
                // PACING: on this tick — already off the request path, once a
                // minute — and a thirtieth of the namespaces at a time (~330
                // one-repository walks a minute at 10,000 tenants).
                reconcile_tick(&tick, FLOW_FLUSH_INTERVAL);
                // **The dashboard's fold, taken HERE and not on the band's own
                // tick.** It walks every namespace and reads the control
                // document; the band re-samples every two seconds and is
                // documented as reading a few atomics and nothing else. Once a
                // minute is also the right rate for a row a human reads.
                *band.lock().expect("the flow band") = Some(
                    tick.flow()
                        .fleet(gunnar_core::now_ms(), crate::dashboard::SILENT_AFTER_DAYS),
                );
                written
            })
            .await
            .unwrap_or_else(|err| FlowFlush::Failed(err.to_string()));
            match written {
                FlowFlush::Wrote(0) => {}
                // The release window of a runtime growth: nothing lost, nothing to say above
                // trace. The counters stay in RAM and the first tick after the remount writes.
                FlowFlush::Held => tracing::trace!(
                    target: "gunnar::store",
                    event = "flow.flush_held",
                    "the transaction log is released for a data-set growth; the flow ledger waits for the remount"
                ),
                FlowFlush::Wrote(rows) => tracing::debug!(
                    target: "gunnar::store",
                    event = "flow.flushed",
                    rows,
                    "wrote {rows} namespace-day flow row(s)"
                ),
                FlowFlush::Failed(err) => tracing::warn!(
                    target: "gunnar::store",
                    event = "flow.flush_failed",
                    %err,
                    "the flow ledger could not be written; the counters are still in memory and \
                     the next tick will try again"
                ),
            }
        }
    })
}

/// **One tick of the usage ledger's reconcile pass, a miss counted** (A28 M17): the pass
/// that could not run is counted on the server's counters (`usage_reconcile_missed`, on
/// `/metrics` and in the heartbeat) and warned, never dropped; the next tick tries again.
pub fn reconcile_tick(stores: &ServerStores, tick: std::time::Duration) -> Option<(usize, usize)> {
    count_reconcile(
        stores.usage_ledger().reconcile_step(stores, tick),
        stores.metrics(),
    )
}

/// Pure but for the counter: **the reconcile pass's outcome, a miss counted and warned**.
pub fn count_reconcile(
    outcome: gunnar_core::Result<(usize, usize)>,
    metrics: &gunnar_core::observe::ServerMetrics,
) -> Option<(usize, usize)> {
    match outcome {
        Ok(done) => Some(done),
        Err(err) => {
            metrics.add_usage_reconcile_missed(1);
            tracing::warn!(
                target: "gunnar::store",
                event = "usage.reconcile_missed",
                %err,
                missed = metrics.snapshot().usage_reconcile_missed,
                "the usage ledger's reconcile pass could not run ({err}): nothing was re-measured \
                 this tick; counted as usage_reconcile_missed, and the next tick tries again"
            );
            None
        }
    }
}

/// Bytes under `dir`, walked (not `du`, not estimated). `0` for a missing dir.
pub fn dir_bytes(dir: &Path) -> u64 {
    fn walk(p: &Path, acc: &mut u64) {
        let Ok(rd) = std::fs::read_dir(p) else { return };
        for e in rd.flatten() {
            let Ok(m) = e.metadata() else { continue };
            if m.is_dir() {
                walk(&e.path(), acc);
            } else {
                *acc += m.len();
            }
        }
    }
    let mut n = 0;
    walk(dir, &mut n);
    n
}

/// **The newest mtime anywhere under `dir`, as unix milliseconds.**
///
/// The directory's own mtime is not enough: a store is a tree, and writing a
/// pack inside `objects/` does not touch the top. This is what tells an
/// operator on a shell-less appliance whether an orphan is Friday's wreckage
/// or something a push is still writing to — which is the difference between
/// `repo reclaim` and leaving it alone.
///
/// `0` when nothing can be read, which is a fact ("no timestamp") rather than
/// a guess at one.
pub fn newest_mtime_unix_ms(dir: &Path) -> i64 {
    fn walk(p: &Path, acc: &mut i64) {
        let Ok(rd) = std::fs::read_dir(p) else { return };
        for e in rd.flatten() {
            let Ok(m) = e.metadata() else { continue };
            if let Ok(at) = m.modified() {
                if let Ok(since) = at.duration_since(std::time::UNIX_EPOCH) {
                    *acc = (*acc).max(since.as_millis() as i64);
                }
            }
            if m.is_dir() {
                walk(&e.path(), acc);
            }
        }
    }
    let mut newest = std::fs::metadata(dir)
        .and_then(|m| m.modified())
        .ok()
        .and_then(|at| at.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    walk(dir, &mut newest);
    newest
}

/// **Is the persisted role a STANDBY?** Read from `replication.json` once, at
/// open, to decide whether the box comes up frozen.
///
/// Behind the `replication` feature because the role lives in the replication
/// config; a build without it has no way to become a standby and is always a
/// writable primary. A missing or malformed file reads as PRIMARY — the same
/// fail-open chosen for a torn `replication.json`, because a box that
/// refused every push over a config it could not read would be a worse outage
/// than one that served.
#[cfg(feature = "replication")]
fn replication_role_at_open(root: &Path) -> crate::mirror::Role {
    crate::mirror::ReplicationConfig::load(root)
        .map(|c| c.role)
        .unwrap_or_default()
}

/// The role a build without replication always has: a writable primary. The
/// type is a private stand-in with the two questions the open path asks.
#[cfg(not(feature = "replication"))]
fn replication_role_at_open(_root: &Path) -> PrimaryOnly {
    PrimaryOnly
}

#[cfg(not(feature = "replication"))]
struct PrimaryOnly;

#[cfg(not(feature = "replication"))]
impl PrimaryOnly {
    fn refuses_client_writes(&self) -> bool {
        false
    }
    fn is_browse(&self) -> bool {
        false
    }
}

#[cfg(test)]
mod tombstone_tests {
    use super::*;

    /// **The old free-text tombstone must not read as a soft delete.**
    ///
    /// An appliance upgraded in place has files on disk that this code did not
    /// write. Reading one of those as `Deleted` would put a *deadline* on bytes
    /// that are OWED to an operator, and the sweep would then quietly do the
    /// `remove_dir_all` that already failed once — unattended, at 3 a.m.
    ///
    /// Seen red by making `parse` fall back to `Deleted`: this goes red on the
    /// reason and on `is_expired_at`.
    #[test]
    fn a_free_text_tombstone_reads_as_a_failed_reclaim_and_never_expires() {
        let old = "delete asked at unix_ms 1756500000000, reclaim failed: Permission denied\n";
        let record = Tombstone::parse(old);
        assert_eq!(record.reason, TombstoneReason::ReclaimFailed);
        assert_eq!(record.purge_after_unix_ms, None);
        assert!(record.note.contains("Permission denied"));
        // The load-bearing half: no deadline means the sweep never touches it,
        // whatever "now" is.
        assert!(!record.is_expired_at(i64::MAX));
        assert_eq!(record.seconds_left(0), None);
    }

    /// A soft delete round-trips through JSON with its deadline and its row.
    #[test]
    fn a_soft_delete_round_trips_and_expires_exactly_at_its_deadline() {
        let record = Tombstone {
            deleted_at_unix_ms: 1_000,
            purge_after_unix_ms: Some(3_000),
            principal_id: "alice".into(),
            reason: TombstoneReason::Deleted,
            repo: None,
            note: "soft-deleted".into(),
        };
        let text = serde_json::to_string(&record).unwrap();
        let back = Tombstone::parse(&text);
        assert_eq!(back, record);
        assert!(!back.is_expired_at(2_999), "one ms early is not expired");
        assert!(back.is_expired_at(3_000), "the deadline itself expires");
        assert!(back.is_expired_at(3_001));
        assert_eq!(back.seconds_left(1_000), Some(2));
        assert_eq!(back.seconds_left(9_999), Some(0), "never negative");
    }

    /// The two reasons print differently, because they mean different things
    /// and an operator acts on them differently.
    #[test]
    fn the_two_reasons_do_not_share_a_word() {
        assert_ne!(
            TombstoneReason::Deleted.as_str(),
            TombstoneReason::ReclaimFailed.as_str()
        );
    }
}

/// One `statvfs` of a path gunnar was TOLD to use.
///
/// **Never a number that was not read.** A failed measurement leaves
/// `measured` false and both byte fields at zero with the OS's reason in
/// `error`; a renderer that prints `0 B free` for a filesystem nobody could
/// measure is how an operator is sent to look for a full disk that is not
/// full.
pub(crate) fn filesystem_use(what: &str, path: &Path) -> gunnar_proto::FilesystemUse {
    let mut row = gunnar_proto::FilesystemUse {
        what: what.to_owned(),
        path: path.display().to_string(),
        ..Default::default()
    };
    match rustix::fs::statvfs(path) {
        Ok(vfs) => {
            let frsize = if vfs.f_frsize == 0 {
                vfs.f_bsize
            } else {
                vfs.f_frsize
            };
            row.measured = true;
            row.total_bytes = vfs.f_blocks.saturating_mul(frsize);
            row.free_bytes = vfs.f_bavail.saturating_mul(frsize);
        }
        Err(err) => row.error = err.to_string(),
    }
    row
}

#[cfg(test)]
mod flow_flush_tests {
    use super::*;

    /// ★ **A reconcile pass that cannot run is counted, published and warned, never dropped**
    /// (A28 M17). Before, `reconcile_step` answered `(0, 0)` for a catalog it could not read,
    /// and the flush tick threw the answer away (`let _ =`): a ledger that stopped
    /// re-measuring looked exactly like one that found nothing to correct.
    #[test]
    fn a_reconcile_pass_that_cannot_run_is_counted_on_the_servers_counters() {
        let metrics = gunnar_core::observe::ServerMetrics::new();
        assert_eq!(count_reconcile(Ok((3, 1)), &metrics), Some((3, 1)));
        assert_eq!(
            metrics.snapshot().usage_reconcile_missed,
            0,
            "a pass that ran is no miss"
        );
        let miss = Err(gunnar_core::Error::NoSuchNamespace(
            "catalog unreadable".into(),
        ));
        assert_eq!(count_reconcile(miss, &metrics), None);
        assert_eq!(metrics.snapshot().usage_reconcile_missed, 1);
        assert!(
            metrics
                .snapshot()
                .render_prometheus()
                .contains("gunnar_usage_reconcile_missed 1\n"),
            "the miss is on /metrics: {}",
            metrics.snapshot().render_prometheus()
        );
    }

    /// ★ The tick runs on the stores' OWN counters, the set `Server::start` takes: a healthy
    /// pass over a fresh box counts no miss.
    #[test]
    fn the_reconcile_tick_counts_on_the_stores_one_set_of_counters() {
        let dir = tempfile::tempdir().unwrap();
        let stores = ServerStores::open(dir.path()).unwrap();
        assert_eq!(reconcile_tick(&stores, FLOW_FLUSH_INTERVAL), Some((0, 0)));
        assert_eq!(stores.metrics().snapshot().usage_reconcile_missed, 0);
    }

    /// **A runtime growth's release window is a quiet hold, and nothing is lost across it.**
    ///
    /// Found by the lifecycle lane 2026-09-15: `flow.flush_failed` every minute while the txlog
    /// was released. Yesterday's counters are the ones at risk — the roll moves them out of the
    /// live slot into `completed`, and a flush whose write was refused used to drop them there.
    #[test]
    fn a_released_txlog_holds_the_flush_and_the_rolled_day_lands_after_reacquire() {
        let dir = tempfile::tempdir().unwrap();
        let stores = ServerStores::open(dir.path()).unwrap();
        let now = gunnar_core::now_ms();
        let yesterday = now - 86_400_000;
        let push = gunnar_core::flow::Transfer {
            bytes_in: 42,
            pushed: true,
            principal: "p".into(),
            ..Default::default()
        };
        stores.flow().record("ns", yesterday, &push);

        stores.txlog().release().unwrap();
        // The red the tick used to log: the write itself is refused while released.
        assert!(
            stores.flow().flush(now).is_err(),
            "a flush wrote through a released transaction log"
        );
        assert_eq!(flow_flush_tick(&stores), FlowFlush::Held);
        assert_eq!(
            flow_flush_tick(&stores),
            FlowFlush::Held,
            "still held, every tick of the window"
        );

        stores.txlog().reacquire().unwrap();
        assert_eq!(
            flow_flush_tick(&stores),
            FlowFlush::Wrote(1),
            "yesterday's row, re-parked by the refused flush, is written after the remount"
        );
        assert_eq!(
            flow_flush_tick(&stores),
            FlowFlush::Wrote(0),
            "and written once"
        );
    }
}

/// **Frame the site's WireGuard hub declaration into this primary's log, once per change.**
///
/// `wg/hub.conf` is public site data that `korp-installer` lays into `/data` through the ESP
/// data stage on a fresh install and that nothing in gunnar ever writes. It therefore had no
/// sink call site, the twin never held it, and `gunnar control twin replay` brought a main
/// back with no way of reaching the tunnel — the D6 failure again, in the one class where the
/// symptom is "nobody can reach the box at all".
///
/// **Best effort, and deliberately so.** This runs on the start-up path of a box whose tunnel
/// is already configured — the file on disk is the truth either way, and refusing to serve
/// because a 1 KiB document could not be framed would turn a replication gap into an outage.
/// A failure is one WARN naming the path, at the place an operator is already looking.
///
/// The machine's own `wg/identity.key` and `wg/identity.pub` are NOT touched here and are
/// refused by `gunnar_core::txlog::is_per_box_path`: a private key on the twin is a second
/// place to steal the site's network identity from, and a restored main carrying the dead
/// box's identity would claim a tunnel the hub registered to a key that no longer exists.
fn adopt_hub_conf(root: &Path, txlog: &Arc<gunnar_core::txlog::Log>) {
    use gunnar_core::txlog::{HUB_CONF_FRAMED_REL, HUB_CONF_REL};
    let path = root.join(HUB_CONF_REL);
    let bytes = match std::fs::read(&path) {
        Ok(b) => b,
        // No declaration is the ordinary case on every box that is not an appliance, and on
        // an appliance before the ceremony. Nothing to say.
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return,
        Err(err) => {
            tracing::warn!(
                target: gunnar_core::observe::TARGET_TWIN,
                path = %path.display(),
                %err,
                "the hub declaration could not be read, so it is NOT in this box's log and a twin will not hold it"
            );
            return;
        }
    };
    // The ONE place that decides what a digest of something is.
    let digest = gunnar_core::provenance::digest_bytes(&bytes);
    let mark = root.join(HUB_CONF_FRAMED_REL);
    // Already framed, unchanged: the log carries these bytes. One `read_to_string` per boot.
    if std::fs::read_to_string(&mark)
        .map(|s| s.trim() == digest)
        .unwrap_or(false)
    {
        return;
    }
    let framed = txlog.append(
        gunnar_core::txlog::Kind::File,
        &gunnar_core::txlog::Meta::File {
            path: HUB_CONF_REL.to_string(),
            deleted: false,
        },
        &bytes,
    );
    match framed {
        Ok(seq) => {
            // The mark is written AFTER the frame is durable. The other order would let a
            // crash between them leave a box claiming to have framed what it had not — and
            // then never framing it, which is exactly the silent loss being fixed. This order
            // costs at worst one duplicate frame of the same bytes after a crash.
            if let Err(err) = crate::durable::write_atomic_unframed(
                mark.parent().unwrap_or(root),
                HUB_CONF_FRAMED_REL
                    .rsplit('/')
                    .next()
                    .unwrap_or(HUB_CONF_FRAMED_REL),
                format!("{digest}\n").as_bytes(),
            ) {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "the hub declaration is framed at seq {seq}; the note saying so was not written, so the next boot frames it again");
            }
            tracing::info!(
                target: gunnar_core::observe::TARGET_TWIN,
                seq,
                path = %path.display(),
                "the site's hub declaration is a transaction: the twin holds it and a restore brings it back"
            );
        }
        Err(err) => tracing::warn!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            %err,
            "the hub declaration could not be framed, so a twin will not hold it and a restored main would come up with no way onto the tunnel"
        ),
    }
}

/// **Mark a namespace in the usage ledger when the operation ENDS** — on
/// every path out, errors included. Marking at the start would let a settle
/// in the middle measure the half-done state and clear the mark; at the end, a
/// mid-way measurement is simply taken again.
struct UsageMark<'a> {
    ledger: &'a crate::usage_ledger::UsageLedger,
    namespace: String,
}

impl<'a> UsageMark<'a> {
    fn namespace(ledger: &'a crate::usage_ledger::UsageLedger, namespace: &str) -> Self {
        UsageMark {
            ledger,
            namespace: namespace.to_owned(),
        }
    }
}

impl Drop for UsageMark<'_> {
    fn drop(&mut self) {
        self.ledger.mark_namespace(&self.namespace);
    }
}

/// **The named sealed slot's row** (2026-10-09): the secret `slot/<name>`,
/// holding `gsl1 | u16 owner length | owner | value` — sealed as one, so the
/// owner cannot be moved off its value.
#[cfg(feature = "metered")]
pub mod sealed_slot {
    use anyhow::{bail, Result};

    /// Every slot's secret key starts so; no M5 secret does.
    pub const PREFIX: &str = "slot/";
    const MAGIC: &[u8; 4] = b"gsl1";

    pub fn key(name: &str) -> String {
        format!("{PREFIX}{name}")
    }

    pub fn encode(owner: &str, value: &[u8]) -> Vec<u8> {
        let mut out = Vec::with_capacity(6 + owner.len() + value.len());
        out.extend_from_slice(MAGIC);
        out.extend_from_slice(&(owner.len() as u16).to_be_bytes());
        out.extend_from_slice(owner.as_bytes());
        out.extend_from_slice(value);
        out
    }

    pub fn decode(b: &[u8]) -> Result<(String, Vec<u8>)> {
        if b.len() < 6 || &b[..4] != MAGIC {
            bail!("a sealed slot's row is not gsl1");
        }
        let n = u16::from_be_bytes([b[4], b[5]]) as usize;
        if b.len() < 6 + n {
            bail!("a sealed slot's row is cut short");
        }
        Ok((
            String::from_utf8(b[6..6 + n].to_vec())?,
            b[6 + n..].to_vec(),
        ))
    }
}

/// ★ **A re-laid main mints no ticket key of its own before its twin gives the old one back**
/// (gunnar.rs audit round 2, RCH-TK). After `estate recharge main` the ONE box keeps the
/// site's appliance key: the dead main's, which the twin's sealed secrets restore. A box that
/// awaits its twin (`pairing/recharge_from`, [`crate::pairing::Hold::Awaiting`]) or is being
/// restored and minted a key first — the installer's offer and self-registration both ask —
/// would sign every ticket with a key the ONE box refuses, and the write itself would end
/// the box's freshness before the restore.
#[cfg(all(test, feature = "metered"))]
mod ticket_key_recharge_tests {
    use super::*;

    #[test]
    fn a_main_awaiting_its_twin_mints_no_ticket_key_and_reads_the_restored_one_after() {
        use gunnar_store::SecretStore as _;
        let dir = tempfile::tempdir().unwrap();
        let stores = ServerStores::open(dir.path()).unwrap();
        let marker = dir.path().join(crate::pairing::RECHARGE_REL);
        std::fs::create_dir_all(marker.parent().unwrap()).unwrap();
        std::fs::write(&marker, "10.13.8.3:50051\n").unwrap();

        let e = stores
            .actor_ticket_key()
            .err()
            .expect("no key minted while the box awaits its twin");
        assert!(format!("{e:#}").contains("awaits its twin"), "{e:#}");
        assert!(
            stores
                .secrets()
                .unwrap()
                .get(gunnar_store::ACTOR_TICKET_KEY)
                .unwrap()
                .is_none(),
            "nothing was written"
        );

        // The restore brought the dead main's key back and the marker went: it is read, not minted.
        std::fs::remove_file(&marker).unwrap();
        let k = stores
            .actor_ticket_key()
            .expect("restored (or never recharged): the key is there or minted");
        assert_eq!(
            stores.actor_ticket_key().unwrap().public_key_hex(),
            k.public_key_hex()
        );
    }
}

/// The jera job that flushes the last-seen record (A32-14).
pub const ACTIVITY_FLUSH: &str = "principal-activity";

/// **The last-seen record's flush, every `every`, as a jera job** (A32-14): the raw
/// `principal-activity` thread ended at a panic without a word and was never stopped. It holds the
/// record weakly and ends with it (the record's drop writes the last stamps); a flush that fails
/// keeps the rows in memory and is said.
pub fn start_activity_flush(
    activity: &Arc<gunnar_core::PrincipalActivity>,
    every: std::time::Duration,
) -> Result<Arc<gunnar_common::jobs::Job>, String> {
    let weak = Arc::downgrade(activity);
    let mut first = true;
    let say: gunnar_common::jobs::Say = Arc::new(|l: &str| tracing::error!("{l}"));
    gunnar_common::jobs::start(ACTIVITY_FLUSH, every, say, move || {
        // The first look only waits, as the thread's first sleep did.
        if std::mem::take(&mut first) {
            return gunnar_common::jobs::Round::Again;
        }
        let Some(activity) = weak.upgrade() else {
            return gunnar_common::jobs::Round::Done;
        };
        if let Err(err) = activity.flush() {
            tracing::warn!(error = %err, "the last-seen rows could not be written; they are kept in memory and tried again");
        }
        gunnar_common::jobs::Round::Again
    })
}

#[cfg(test)]
mod activity_flush_tests {
    use super::*;

    /// ★ **The last-seen flush is a jera job that writes while the record lives and ends with it.**
    #[test]
    fn the_activity_flush_is_a_jera_job_that_ends_with_the_record() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        let db = gunnar_core::control::open_control_with_db(&path)
            .expect("control store")
            .2;
        let a = Arc::new(gunnar_core::PrincipalActivity::open(
            Arc::clone(&db),
            None,
            None,
        ));
        let job = start_activity_flush(&a, std::time::Duration::from_millis(10))
            .expect("the flush starts as a jera job");
        assert_eq!(job.kind(), ACTIVITY_FLUSH);
        a.signed_in("kim", 7);
        let t = std::time::Instant::now();
        let flushed = || {
            gunnar_core::PrincipalActivity::open(Arc::clone(&db), None, None)
                .seen("kim")
                .signed_in
                == Some(7)
        };
        while !flushed() && t.elapsed() < std::time::Duration::from_secs(10) {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(
            flushed(),
            "the job flushed the stamp while the record lives"
        );
        drop(a);
        let t = std::time::Instant::now();
        while !job.is_stopped() && t.elapsed() < std::time::Duration::from_secs(10) {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(job.is_stopped(), "the job ended with the record");
    }
}