gunnar-core 3.0.0

//! **Where the control plane's own state lives** — the catalog, the identity
//! table, and every document [`crate::doc`] puts behind a name.
//!
//! # Why this is not the object store any more (the certificate-identity plan)
//!
//! Until 2026-08-09 a control document was a blob in the repository object
//! store with a reference pointing at it — gunnar's own `ObjectStore` and
//! `RefStore` traits, both since deleted in favour of `GitOps`: git's own
//! trick, and defensible while the only durable backend gunnar had *was* a git
//! directory. It stopped being defensible when the storage engine landed
//! (the storage restart). That engine takes objects **as packs** — `GitOps::put_pack` is
//! where the closure check and the durability ordering live — and it refuses a
//! single-object write by design. So `principal add` and `namespace create`
//! both died before they reached any of their own logic, with
//!
//! ```text
//!   the znippy-git store takes objects as packs, not one at a time
//! ```
//!
//! The fix is not a pack wrapper around a 200-byte JSON blob. User records were
//! never git objects for a *reason*; they were git objects because the control
//! plane predates anywhere better. This is that better place: an embedded
//! key/value store with real transactions, which is what a compare-and-swap on
//! a document wanted all along.
//!
//! redb adds nothing to the appliance that was not already there — it is the
//! znippy engine's own tail — and it has no build script beyond a version probe
//! and no dependencies of its own.
//!
//! # What is stored, and in which table
//!
//! ```text
//!   documents          name            → 32-byte blob id      the current version
//!   blobs              32-byte id      → bytes                content-addressed
//!   document_history   (name, seq)     → millis ‖ blob id     append-only
//!   secrets            key             → tag ‖ sealed bytes   crate::secrets
//! ```
//!
//! The fourth table arrived on 2026-08-10 and finished the migration this one
//! started: the SSH host key and the issuer private key were still in an Arrow
//! log inside `control.git`, which is what kept `gunnar-server` opening a
//! repository — and therefore needing a storage engine — in order to boot. It
//! is a *table* here rather than a document because a secret is not a document:
//! there is no compare-and-swap on it, no history worth keeping (a superseded
//! host key is the defect, not the audit trail) and no content addressing to
//! want. See [`crate::secrets`], which owns everything about its shape.
//!
//! **Content addressing is kept**, and not out of nostalgia for git. Three
//! properties of [`crate::doc`] rest on it and would each have to be rebuilt
//! without it: a shard that did not change keeps its id, so a document that did
//! not change hashes to the same bytes and a no-op update is visible as *the
//! pointer did not move* rather than as a new version; a losing writer's shards
//! are simply unreferenced rather than needing a rollback; and a reader that
//! follows a document to its shards can never find a leaf that was rewritten
//! underneath it.
//!
//! # The two things moving off git objects cost (the certificate-identity plan)
//!
//! * **Replication.** Metadata used to replicate *as a repository*, by the same
//!   mechanism as everything else: `control.git` mirrored, and the catalog and
//!   identity table came with it. It does not any more. `<data-root>/control.redb`
//!   is a file a standby has to be given some other way, and there is no
//!   mechanism here that gives it one. **Named, not rebuilt.**
//! * **History.** A ref update per change was a free audit trail: every previous
//!   version of the identity table stayed in the object store with the reflog
//!   ordering them. That is *not* free on a key/value store, so it is explicit:
//!   [`DOC_HISTORY`] is appended in the same write transaction as the pointer
//!   move, and [`ControlStore::history`] reads it back. Rebuilt, deliberately —
//!   an ACL whose changes cannot be listed is an ACL nobody can audit, and the
//!   whole table is one B-tree append per change.
//!
//! Nothing prunes either table. Old document versions and superseded shards
//! accumulate exactly as loose objects did, which is the same bargain git makes
//! and the same one a `gunnar control gc` would have to unmake.

use std::path::Path;
use std::sync::Arc;

use redb::ReadableTable;

use crate::error::{Error, Result};

/// The current version of every named document.
const DOCUMENTS: redb::TableDefinition<'static, &str, &[u8]> =
    redb::TableDefinition::new("documents");

/// Content-addressed bytes: documents themselves, and the shard blobs they
/// point at.
const BLOBS: redb::TableDefinition<'static, &[u8], &[u8]> = redb::TableDefinition::new("blobs");

/// **The audit trail** (the certificate-identity plan): every version a
/// document has ever had, in the order it had them, written in the same
/// transaction as the pointer move.
///
/// Keyed `(document name, sequence)` so one document's history is one range
/// scan and two documents cannot interleave. The value is the commit time in
/// milliseconds since the epoch, big-endian, followed by the blob id.
const DOC_HISTORY: redb::TableDefinition<'static, (&str, u64), &[u8]> =
    redb::TableDefinition::new("document_history");

/// **What this box has applied, by sequence number** — per box, never framed, never in a
/// snapshot or an export (perf twin-apply-fsync, 2026-10-06). One row today: [`TWIN_APPLIED`].
const MARKS: redb::TableDefinition<'static, &str, u64> = redb::TableDefinition::new("box_marks");

/// **When each principal last signed in and last changed something** (owner,
/// 2026-10-21) — principal id → its stamps ([`crate::activity`], JSON). Per box
/// like [`MARKS`]: never framed, never in a snapshot or an export, and never in
/// the identity document. Written by the activity flusher only, one transaction
/// per flush that moved something.
const PRINCIPAL_ACTIVITY: redb::TableDefinition<'static, &str, &[u8]> =
    redb::TableDefinition::new("principal_activity");

/// **The last transaction-log entry whose control-plane writes are in this database** — a
/// twin's applied watermark. Committed in the SAME redb transaction as the writes it vouches
/// for ([`ControlDb::commit_group`]), so after a crash the database is exactly the state through
/// the marker, and the entries after it are re-applied in order from there. A twin used to re-apply
/// only its log's head, which is exact only when every entry is made durable on its own; a
/// group commit needs the marker.
pub const TWIN_APPLIED: &str = "twin.applied";

/// **What a write closure decided** ([`ControlDb::write`]): commit what it wrote, or discard
/// the transaction — which a closure may only do when it wrote NOTHING (a lost compare-and-swap,
/// a secret that already exists), because inside a group ([`ControlDb::begin_group`]) a discard
/// cannot take back what the transaction already holds.
pub(crate) enum Txn<T> {
    /// Keep the writes.
    Commit(T),
    /// Keep the writes WITHOUT an fsync of their own: durable with the next durable commit
    /// ([`ControlStore::put_blob_unsynced`]).
    CommitUnsynced(T),
    /// Nothing was written; nothing to keep.
    Discard(T),
}

/// **One redb write transaction shared by a run of writes on ONE thread** (perf
/// twin-apply-fsync, 2026-10-06): a twin takes a whole group of log entries into it and commits
/// it once — each B-tree page it touches is written once, and the group costs one fsync.
struct Group {
    owner: std::thread::ThreadId,
    txn: redb::WriteTransaction,
    /// A write committed into it.
    dirty: bool,
    /// A write failed half-way: the group is aborted whole at its end.
    poisoned: bool,
}

/// How many bytes a [`BlobId`] is.
pub const BLOB_ID_LEN: usize = 32;

/// A content address: sha-256 over the stored bytes.
///
/// Not a `gix_hash::ObjectId`. The old id was a *git* object id because the
/// bytes were in a git object database and git had to be able to name them.
/// Nothing here is git-visible, so the id is the hash and nothing else — no
/// object kind, no `blob <len>\0` header, and no hash-kind negotiation with a
/// repository that is not involved.
#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct BlobId([u8; BLOB_ID_LEN]);

impl BlobId {
    /// The address of these bytes.
    pub fn of(bytes: &[u8]) -> Self {
        use sha2::Digest;
        let mut hasher = sha2::Sha256::new();
        hasher.update(bytes);
        BlobId(hasher.finalize().into())
    }

    /// The raw address.
    pub fn as_bytes(&self) -> &[u8; BLOB_ID_LEN] {
        &self.0
    }

    /// Parse the hex form. `None` if it is not [`BLOB_ID_LEN`] bytes of hex.
    pub fn from_hex(text: &str) -> Option<Self> {
        let mut out = [0u8; BLOB_ID_LEN];
        hex::decode_to_slice(text, &mut out).ok()?;
        Some(BlobId(out))
    }

    /// Rebuild from raw bytes, as they come back out of a table.
    fn from_slice(bytes: &[u8]) -> Option<Self> {
        Some(BlobId(bytes.try_into().ok()?))
    }
}

impl std::fmt::Display for BlobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&hex::encode(self.0))
    }
}

impl std::fmt::Debug for BlobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "BlobId({self})")
    }
}

/// One version a document has had, out of [`ControlStore::history`].
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct DocVersion {
    /// Its position in the document's history, from 1.
    pub seq: u64,
    /// When it was committed, in milliseconds since the unix epoch.
    pub at: crate::model::UnixMillis,
    /// The version itself, still readable through
    /// [`ControlStore::get_blob`] — nothing prunes the blob table.
    pub blob: BlobId,
}

/// **Where a control document is kept.**
///
/// A trait rather than the concrete store, for one reason, and it is measure,
/// never assert:
/// `tests/control_cost.rs` proves the compare-and-swap by running the same
/// concurrency test against a store that *ignores the expectation*, in the same
/// run, and asserting that acknowledged writes are lost. A guard that cannot be
/// broken is a guard nobody has seen fail. The counting decorator that measures
/// what a connection costs lives on the same seam.
pub trait ControlStore: Send + Sync + 'static {
    /// Store these bytes and return their address. Idempotent: storing the same
    /// bytes twice is one row and the same id.
    fn put_blob(&self, bytes: &[u8]) -> Result<BlobId>;

    /// **[`ControlStore::put_blob`] for a blob that only a LATER [`ControlStore::commit_doc`]
    /// makes part of any state** (perf twin-apply-fsync, 2026-10-06): the store may keep it
    /// without making it durable on its own, because the pointer move that publishes it is
    /// durable and persists it — redb's durable commit persists every earlier commit. Until then
    /// it is an unreferenced, content-addressed row nothing can reach, and a crash that takes it
    /// takes nothing anybody could see. The default is the durable `put_blob`.
    ///
    /// The primary's tee uses it for a document's shards. A replay that re-applies only its
    /// head entry after a crash (a box being restored) must NOT: its blob frames are durable
    /// one by one and the document that publishes them may be several frames on.
    fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
        self.put_blob(bytes)
    }

    /// The bytes at this address, or `None` if nothing was ever stored there.
    fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>>;

    /// Which version `name` is at, or `None` for a document that has never been
    /// written.
    ///
    /// **`None` is a legitimate state and a fault is not.** A read that could
    /// not distinguish them would report an unreadable identity table as *no
    /// principals are registered*, which is the failure that opens a server up.
    fn read_doc(&self, name: &str) -> Result<Option<BlobId>>;

    /// **The compare-and-swap.** Move `name` from `expected` to `new`, and
    /// append the new version to the audit trail, atomically.
    ///
    /// `Ok(false)` means the document was not at `expected` — another writer
    /// committed in between — and the caller must re-read and re-apply. It is
    /// deliberately not an error: losing a race is an ordinary outcome of a
    /// contended document and a caller that retried on `Err` would also retry
    /// on a broken disk.
    fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool>;

    /// Every version `name` has had, oldest first (the control plane's audit trail).
    fn history(&self, name: &str) -> Result<Vec<DocVersion>>;
}

/// **The RAM copy of the control database**, taken by [`ControlDb::suspend`]
/// the instant before `control.redb`'s file is closed and held until
/// [`ControlDb::resume`] reopens it. Every current document pointer, every
/// blob (current versions, their shards, and the history — the blobs table
/// whole, because a shard is reached from a document by id and nothing
/// short of the table says which ids a document reaches), and every sealed
/// secret row. Reads are answered from it verbatim; writes are refused by
/// name ([`Error::OffStripe`]).
#[derive(Debug, Clone)]
pub struct RamCopy {
    /// Why the store is off the stripe — the growth's phase, in words.
    pub why: String,
    docs: std::collections::BTreeMap<String, BlobId>,
    blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
    secrets: std::collections::BTreeMap<String, Vec<u8>>,
    /// When it was taken, unix ms.
    pub taken_unix_ms: crate::model::UnixMillis,
}

impl RamCopy {
    /// How many bytes the copy holds (blobs and secret rows).
    pub fn bytes(&self) -> u64 {
        self.blobs.values().map(|b| b.len() as u64).sum::<u64>()
            + self.secrets.values().map(|b| b.len() as u64).sum::<u64>()
    }
    /// How many documents have a current version.
    pub fn documents(&self) -> usize {
        self.docs.len()
    }
    /// The current pointer of every document, name → blob id hex — what
    /// [`ControlDb::resume`] compares the disk against.
    pub fn pointers(&self) -> Vec<(String, String)> {
        self.docs
            .iter()
            .map(|(k, v)| (k.clone(), v.to_string()))
            .collect()
    }
    /// A secret's raw row (tag ‖ sealed bytes), for the secret store.
    pub(crate) fn secret_row(&self, key: &str) -> Option<&[u8]> {
        self.secrets.get(key).map(|v| v.as_slice())
    }
}

/// **The control database's current state, read without taking it off the stripe**: every
/// document pointer, every blob, every secret row (`None` = a tombstone; values are the SEALED
/// bytes). What a twin snapshot carries and what a restored box re-seals.
#[derive(Debug, Clone, Default)]
pub struct ControlExport {
    /// Document name → current blob.
    pub docs: std::collections::BTreeMap<String, BlobId>,
    /// Every blob, current and historical.
    pub blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
    /// Secret key → sealed value, or `None` for a tombstone.
    pub secrets: std::collections::BTreeMap<String, Option<Vec<u8>>>,
}

impl ControlDb {
    /// [`ControlExport`] of the live database (or of the RAM copy while suspended), in one
    /// read transaction.
    pub fn export(&self) -> Result<ControlExport> {
        let raw = self.read(
            "exporting the control database",
            |txn| {
                let mut docs = std::collections::BTreeMap::new();
                let mut blobs = std::collections::BTreeMap::new();
                let table = txn
                    .open_table(DOCUMENTS)
                    .map_err(|err| faulted("opening the documents table", err))?;
                for entry in table
                    .iter()
                    .map_err(|err| faulted("scanning the documents table", err))?
                {
                    let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
                    if let Some(id) = BlobId::from_slice(v.value()) {
                        docs.insert(k.value().to_owned(), id);
                    }
                }
                let table = txn
                    .open_table(BLOBS)
                    .map_err(|err| faulted("opening the blobs table", err))?;
                for entry in table
                    .iter()
                    .map_err(|err| faulted("scanning the blobs table", err))?
                {
                    let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
                    if let Some(id) = BlobId::from_slice(k.value()) {
                        blobs.insert(id, v.value().to_vec());
                    }
                }
                Ok((docs, blobs, crate::secrets::RedbSecretStore::rows(txn)?))
            },
            |ram| Ok((ram.docs.clone(), ram.blobs.clone(), ram.secrets.clone())),
        )?;
        let (docs, blobs, rows) = raw;
        let mut secrets = std::collections::BTreeMap::new();
        for (key, row) in rows {
            let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &row)
                .map_err(|err| Error::ControlStore(err.to_string()))?;
            secrets.insert(key, value);
        }
        Ok(ControlExport {
            docs,
            blobs,
            secrets,
        })
    }
}

/// One row of [`ControlDb::export_each`], borrowed from the read transaction (or the RAM copy):
/// nothing of the database is gathered to hand it over.
#[derive(Debug, Clone, Copy)]
pub enum ExportRow<'a> {
    /// A blob, current or historical.
    Blob(BlobId, &'a [u8]),
    /// A document's current pointer.
    Doc(&'a str, BlobId),
    /// A secret row: the sealed value, or `None` for a tombstone.
    Secret(&'a str, Option<&'a [u8]>),
}

impl ControlDb {
    /// **[`ControlDb::export`], row by row** — every blob, then every document pointer, then
    /// every secret row, each handed to `row` as it is read, in one read transaction. What a
    /// twin snapshot streams: the database never lies in memory whole. `row`'s error stops the
    /// scan and comes back as `Ok(Err(e))`; the database's own faults are the outer error.
    pub fn export_each<E>(
        &self,
        row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
    ) -> Result<std::result::Result<(), E>> {
        let row = std::cell::RefCell::new(row);
        self.read(
            "exporting the control database row by row",
            |txn| export_txn(txn, &mut **row.borrow_mut()),
            |ram| export_ram(ram, &mut **row.borrow_mut()),
        )
    }

    /// **A consistent view of the whole database, taken NOW and read LATER** (R5.1-A,
    /// 2026-10-07): a redb read transaction (MVCC — it holds the state of this instant while
    /// writers go on), or the RAM copy while the store is off the stripe. What a snapshot takes
    /// under the log's freeze, so it can stream the rows AFTER the freeze is gone.
    ///
    /// Said plainly: while a view lives, redb keeps the pages it reads and the file stays open —
    /// a growth that takes the store off the stripe meanwhile cannot unmount until the view goes.
    pub fn view(&self) -> Result<ControlView> {
        if let Some(copy) = read_lock(&self.ram).as_ref() {
            return Ok(ControlView(View::Ram(Box::new(copy.clone()))));
        }
        let guard = read_lock(&self.live);
        let Some(db) = guard.as_ref() else {
            return Err(self.off_stripe());
        };
        let txn = db
            .begin_read()
            .map_err(|err| faulted("beginning a view of the control database", err))?;
        Ok(ControlView(View::Live(Box::new(txn))))
    }
}

/// See [`ControlDb::view`].
pub struct ControlView(View);

enum View {
    Live(Box<redb::ReadTransaction>),
    Ram(Box<RamCopy>),
}

impl ControlView {
    /// **Every last-seen row at the view's instant** ([`crate::activity`]):
    /// what a snapshot carries. Empty for a database that never kept one and
    /// for the RAM copy.
    pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
        match &self.0 {
            View::Live(txn) => activity_rows(txn),
            View::Ram(_) => Ok(Vec::new()),
        }
    }

    /// [`ControlDb::export_each`] over the state the view was taken at.
    pub fn export_each<E>(
        &self,
        row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
    ) -> Result<std::result::Result<(), E>> {
        match &self.0 {
            View::Live(txn) => export_txn(txn, row),
            View::Ram(ram) => export_ram(ram, row),
        }
    }
}

/// Every row of the last-seen table in `txn`; none when it was never made.
fn activity_rows(txn: &redb::ReadTransaction) -> Result<Vec<(String, Vec<u8>)>> {
    let table = match txn.open_table(PRINCIPAL_ACTIVITY) {
        Ok(t) => t,
        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
        Err(err) => return Err(faulted("opening the last-seen table", err)),
    };
    let mut rows = Vec::new();
    for entry in table
        .iter()
        .map_err(|err| faulted("scanning the last-seen table", err))?
    {
        let (k, v) = entry.map_err(|err| faulted("reading a last-seen row", err))?;
        rows.push((k.value().to_owned(), v.value().to_vec()));
    }
    Ok(rows)
}

fn export_txn<E>(
    txn: &redb::ReadTransaction,
    row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
) -> Result<std::result::Result<(), E>> {
    let table = txn
        .open_table(BLOBS)
        .map_err(|err| faulted("opening the blobs table", err))?;
    for entry in table
        .iter()
        .map_err(|err| faulted("scanning the blobs table", err))?
    {
        let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
        if let Some(id) = BlobId::from_slice(k.value()) {
            if let Err(e) = row(ExportRow::Blob(id, v.value())) {
                return Ok(Err(e));
            }
        }
    }
    let table = txn
        .open_table(DOCUMENTS)
        .map_err(|err| faulted("opening the documents table", err))?;
    for entry in table
        .iter()
        .map_err(|err| faulted("scanning the documents table", err))?
    {
        let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
        if let Some(id) = BlobId::from_slice(v.value()) {
            if let Err(e) = row(ExportRow::Doc(k.value(), id)) {
                return Ok(Err(e));
            }
        }
    }
    for (key, raw) in crate::secrets::RedbSecretStore::rows(txn)? {
        let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &raw)
            .map_err(|err| Error::ControlStore(err.to_string()))?;
        if let Err(e) = row(ExportRow::Secret(&key, value.as_deref())) {
            return Ok(Err(e));
        }
    }
    Ok(Ok(()))
}

fn export_ram<E>(
    ram: &RamCopy,
    row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
) -> Result<std::result::Result<(), E>> {
    for (id, bytes) in &ram.blobs {
        if let Err(e) = row(ExportRow::Blob(*id, bytes)) {
            return Ok(Err(e));
        }
    }
    for (name, id) in &ram.docs {
        if let Err(e) = row(ExportRow::Doc(name, *id)) {
            return Ok(Err(e));
        }
    }
    for (key, raw) in &ram.secrets {
        let value = crate::secrets::RedbSecretStore::decode_row_of(key, raw)
            .map_err(|err| Error::ControlStore(err.to_string()))?;
        if let Err(e) = row(ExportRow::Secret(key, value.as_deref())) {
            return Ok(Err(e));
        }
    }
    Ok(Ok(()))
}

/// The most a RAM copy may weigh before [`ControlDb::suspend`] refuses:
/// the control database is small by design (documents and shards, no git
/// objects), and a copy past this is a sign something else is in it.
pub const RAM_COPY_CEILING_BYTES: u64 = 512 << 20;

/// What [`ControlDb::resume`] found when it compared the disk to the copy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResumeReport {
    /// Documents whose pointer on disk equals the pointer in RAM.
    pub documents_matched: usize,
    /// `name: ram <hex> vs disk <hex|absent>` for every document that
    /// differs — empty is the expected answer, since every write was refused
    /// while the copy stood in.
    pub drift: Vec<String>,
    /// How long the store was off the stripe, ms.
    pub off_stripe_ms: i64,
}

/// **The one `redb::Database` under the control plane — and the door the
/// runtime data-set growth takes it off the stripe through.**
///
/// [`RedbControlStore`] and [`RedbSecretStore`] both hold an `Arc` of this
/// rather than of the database, so that ONE call can close the file under
/// every handle in the process and one call can reopen it, with nothing
/// upstream — `AuthTable`, `RepoCatalog`, the SSH `Control`, the gRPC gate
/// — holding a stale descriptor that would pin the mount. That is the whole
/// mechanism behind `DATA-SET-GROWTH-FLOW.md` G4/T2: `control.redb` lives
/// on the stripe, every SSH authentication reads it at call time, and the
/// stripe has to be unmounted for the members to be grown. So:
///
/// * [`ControlDb::suspend`] takes the write side (waiting for in-flight
///   transactions), copies the tables into a [`RamCopy`], and drops the
///   database — the file is closed and `umount /data` can succeed;
/// * while suspended every read is served from the copy, byte for byte, so
///   `principal_by_fingerprint`, the grant check and the bootstrap state keep
///   answering; every write is [`Error::OffStripe`];
/// * [`ControlDb::resume`] reopens the file, compares every document
///   pointer on disk with the copy (they must agree — no write got through)
///   and drops the copy.
///
/// An in-memory database (tests, `--dry-run`) has no file to close: it is
/// kept, and `suspend` still switches every reader onto the copy and every
/// writer onto the refusal, so the semantics are testable without a disk.
pub struct ControlDb {
    path: Option<std::path::PathBuf>,
    live: std::sync::RwLock<Option<redb::Database>>,
    ram: std::sync::RwLock<Option<RamCopy>>,
    /// The group transaction open on one thread, if any ([`ControlDb::begin_group`]).
    group: std::sync::Mutex<Option<Group>>,
    /// Durable commits (each one fsync) since open ([`ControlDb::durable_commits`]).
    durable_commits: std::sync::atomic::AtomicU64,
}

impl std::fmt::Debug for ControlDb {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ControlDb")
            .field("path", &self.path)
            .field("suspended", &self.is_suspended())
            .finish()
    }
}

fn read_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
    lock.read().unwrap_or_else(|poison| poison.into_inner())
}

fn write_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
    lock.write().unwrap_or_else(|poison| poison.into_inner())
}

impl ControlDb {
    fn new(path: Option<std::path::PathBuf>, db: redb::Database) -> Self {
        ControlDb {
            path,
            live: std::sync::RwLock::new(Some(db)),
            ram: std::sync::RwLock::new(None),
            group: std::sync::Mutex::new(None),
            durable_commits: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// A handle over an in-memory database somebody else prepared — the
    /// secret store's own tests build one without a control store.
    #[cfg(test)]
    pub(crate) fn in_memory_for_tests(db: redb::Database) -> Self {
        Self::new(None, db)
    }

    /// Is the store off the stripe right now?
    pub fn is_suspended(&self) -> bool {
        read_lock(&self.ram).is_some()
    }

    /// The RAM copy, while suspended.
    pub fn ram_copy(&self) -> Option<RamCopy> {
        read_lock(&self.ram).clone()
    }

    /// The refusal every write meets while suspended.
    fn off_stripe(&self) -> Error {
        Error::OffStripe {
            why: read_lock(&self.ram)
                .as_ref()
                .map(|r| r.why.clone())
                .unwrap_or_else(|| "suspended".to_owned()),
        }
    }

    /// Run `f` over a read transaction on the live database, or, while
    /// suspended, `ram` over the copy. The read guard is held for the
    /// transaction's whole life, which is what makes `suspend` wait for it.
    pub(crate) fn read<T>(
        &self,
        what: &str,
        live: impl FnOnce(&redb::ReadTransaction) -> Result<T>,
        ram: impl FnOnce(&RamCopy) -> Result<T>,
    ) -> Result<T> {
        if let Some(copy) = read_lock(&self.ram).as_ref() {
            return ram(copy);
        }
        let guard = read_lock(&self.live);
        let Some(db) = guard.as_ref() else {
            return Err(self.off_stripe());
        };
        let txn = db
            .begin_read()
            .map_err(|err| faulted(&format!("beginning {what}"), err))?;
        live(&txn)
    }

    /// Run `f` over a write transaction on the live database; refused while
    /// suspended. `f` says whether to keep what it wrote ([`Txn`]); the commit is
    /// made here — or, on the thread that holds a group ([`ControlDb::begin_group`]),
    /// left to the group's one commit.
    pub(crate) fn write<T>(
        &self,
        what: &str,
        f: impl FnOnce(&redb::WriteTransaction) -> Result<Txn<T>>,
    ) -> Result<T> {
        if self.is_suspended() {
            return Err(self.off_stripe());
        }
        {
            let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
            if let Some(g) = group
                .as_mut()
                .filter(|g| g.owner == std::thread::current().id())
            {
                return match f(&g.txn) {
                    Ok(Txn::Commit(t) | Txn::CommitUnsynced(t)) => {
                        g.dirty = true;
                        Ok(t)
                    }
                    Ok(Txn::Discard(t)) => Ok(t),
                    Err(err) => {
                        g.poisoned = true;
                        Err(err)
                    }
                };
            }
        }
        let guard = read_lock(&self.live);
        let Some(db) = guard.as_ref() else {
            return Err(self.off_stripe());
        };
        let mut txn = db
            .begin_write()
            .map_err(|err| faulted(&format!("beginning {what}"), err))?;
        match f(&txn)? {
            Txn::CommitUnsynced(t) => {
                txn.set_durability(redb::Durability::None);
                txn.commit()
                    .map_err(|err| faulted(&format!("committing {what}"), err))?;
                Ok(t)
            }
            Txn::Commit(t) => {
                txn.commit()
                    .map_err(|err| faulted(&format!("committing {what}"), err))?;
                self.durable_commits
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                Ok(t)
            }
            // Dropped, not committed: redb rolls it back.
            Txn::Discard(t) => Ok(t),
        }
    }

    /// **Open a group on this thread** (perf twin-apply-fsync, 2026-10-06): every write this
    /// thread makes through the stores over this database goes into ONE transaction until
    /// [`ControlDb::commit_group`] or [`ControlDb::abort_group`]. Writes from other threads wait
    /// for it (redb takes one writer at a time); reads see the state before it. A twin's
    /// follower holds one for the control-plane entries of each group it takes.
    pub fn begin_group(&self) -> Result<()> {
        if self.is_suspended() {
            return Err(self.off_stripe());
        }
        let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
        if group.is_some() {
            return Err(Error::ControlStore(
                "a group transaction is already open on this control store".to_owned(),
            ));
        }
        let guard = read_lock(&self.live);
        let Some(db) = guard.as_ref() else {
            return Err(self.off_stripe());
        };
        let txn = db
            .begin_write()
            .map_err(|err| faulted("beginning a group of writes", err))?;
        *group = Some(Group {
            owner: std::thread::current().id(),
            txn,
            dirty: false,
            poisoned: false,
        });
        Ok(())
    }

    /// **End the group with ONE durable commit, the marker `key` = `seq` inside it** — when it
    /// wrote anything, or `mark` asks for the marker anyway. A group that wrote nothing and is not
    /// asked to mark is dropped (no I/O). A group a write failed in is aborted whole and the
    /// failure answered: none of it is kept. Answers whether it committed.
    pub fn commit_group(&self, key: &str, seq: u64, mark: bool) -> Result<bool> {
        let g = self.take_group()?;
        if g.poisoned {
            drop(g.txn);
            return Err(Error::ControlStore(
                "a write inside the group failed; the whole group was rolled back".to_owned(),
            ));
        }
        if !g.dirty && !mark {
            return Ok(false);
        }
        {
            let mut marks = g
                .txn
                .open_table(MARKS)
                .map_err(|err| faulted("opening the marks table", err))?;
            marks
                .insert(key, seq)
                .map_err(|err| faulted("moving a marker", err))?;
        }
        g.txn
            .commit()
            .map_err(|err| faulted("committing a group of writes", err))?;
        self.durable_commits
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        Ok(true)
    }

    /// **How many durable commits (one fsync each) this database made** since it was opened —
    /// what the I/O tests hold to a bound.
    pub fn durable_commits(&self) -> u64 {
        self.durable_commits
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Drop the group on this thread, keeping none of it.
    pub fn abort_group(&self) {
        let _ = self.take_group();
    }

    fn take_group(&self) -> Result<Group> {
        let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
        match group.take() {
            Some(g) if g.owner == std::thread::current().id() => Ok(g),
            Some(g) => {
                *group = Some(g);
                Err(Error::ControlStore(
                    "the open group transaction belongs to another thread".to_owned(),
                ))
            }
            None => Err(Error::ControlStore(
                "no group transaction is open".to_owned(),
            )),
        }
    }

    /// **Move the marker `key` to `seq`, durably** — inside this thread's group when one is open
    /// (made durable with it), else in a commit of its own (one fsync, and every earlier
    /// non-durable commit with it).
    pub fn set_mark(&self, key: &str, seq: u64) -> Result<()> {
        self.write("a marker", |txn| {
            let mut marks = txn
                .open_table(MARKS)
                .map_err(|err| faulted("opening the marks table", err))?;
            marks
                .insert(key, seq)
                .map_err(|err| faulted("moving a marker", err))?;
            Ok(Txn::Commit(()))
        })
    }

    /// The marker `key`, `None` when it was never set (a box from before markers, or a fresh one).
    pub fn mark(&self, key: &str) -> Result<Option<u64>> {
        self.read(
            "a marker read",
            |txn| {
                let table = match txn.open_table(MARKS) {
                    Ok(t) => t,
                    Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
                    Err(err) => return Err(faulted("opening the marks table", err)),
                };
                let found = table
                    .get(key)
                    .map_err(|err| faulted("reading a marker", err))?;
                Ok(found.map(|v| v.value()))
            },
            |_ram| Ok(None),
        )
    }

    /// **Every principal's last-seen row** ([`crate::activity`]): principal id →
    /// its encoded stamps. Empty for a database that never kept one, and while
    /// the store is off the stripe (the RAM copy carries no such table).
    pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
        self.read("a last-seen read", activity_rows, |_ram| Ok(Vec::new()))
    }

    /// **Write last-seen rows in ONE transaction** (one durable commit for the
    /// whole batch), and the marker `mark` = 1 in the same transaction when one
    /// is named. Refused while the store is off the stripe; the caller keeps
    /// the rows and tries again.
    pub fn put_principal_activity(
        &self,
        rows: &[(String, Vec<u8>)],
        mark: Option<&str>,
    ) -> Result<()> {
        self.write("last-seen rows", |txn| {
            {
                let mut table = txn
                    .open_table(PRINCIPAL_ACTIVITY)
                    .map_err(|err| faulted("opening the last-seen table", err))?;
                for (id, value) in rows {
                    table
                        .insert(id.as_str(), value.as_slice())
                        .map_err(|err| faulted("writing a last-seen row", err))?;
                }
            }
            if let Some(key) = mark {
                let mut marks = txn
                    .open_table(MARKS)
                    .map_err(|err| faulted("opening the marks table", err))?;
                marks
                    .insert(key, 1u64)
                    .map_err(|err| faulted("moving a marker", err))?;
            }
            Ok(Txn::Commit(()))
        })
    }

    /// **Take the store off the stripe.** Copies every table into RAM, then
    /// closes the file (an in-memory database is kept). Refuses when already
    /// suspended, and when the copy would weigh more than
    /// [`RAM_COPY_CEILING_BYTES`] — nothing is changed in that case.
    pub fn suspend(&self, why: &str) -> Result<RamCopy> {
        if self
            .group
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .is_some()
        {
            return Err(Error::ControlStore(
                "a group of writes is open on the control store; it is suspended after the group ends".to_owned(),
            ));
        }
        let mut live = write_lock(&self.live);
        if self.is_suspended() {
            return Err(Error::ControlStore(
                "the control store is already suspended".to_owned(),
            ));
        }
        let Some(db) = live.as_ref() else {
            return Err(Error::ControlStore(
                "the control store has no database to suspend".to_owned(),
            ));
        };
        let txn = db
            .begin_read()
            .map_err(|err| faulted("beginning the suspend snapshot", err))?;
        let mut copy = RamCopy {
            why: why.to_owned(),
            docs: Default::default(),
            blobs: Default::default(),
            secrets: Default::default(),
            taken_unix_ms: crate::model::now_ms(),
        };
        {
            let table = txn
                .open_table(DOCUMENTS)
                .map_err(|err| faulted("opening the documents table", err))?;
            for entry in table
                .iter()
                .map_err(|err| faulted("scanning the documents table", err))?
            {
                let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
                if let Some(id) = BlobId::from_slice(v.value()) {
                    copy.docs.insert(k.value().to_owned(), id);
                }
            }
            let table = txn
                .open_table(BLOBS)
                .map_err(|err| faulted("opening the blobs table", err))?;
            for entry in table
                .iter()
                .map_err(|err| faulted("scanning the blobs table", err))?
            {
                let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
                if let Some(id) = BlobId::from_slice(k.value()) {
                    copy.blobs.insert(id, v.value().to_vec());
                }
            }
            copy.secrets = crate::secrets::RedbSecretStore::rows(&txn)?;
        }
        drop(txn);
        let bytes = copy.bytes();
        if bytes > RAM_COPY_CEILING_BYTES {
            return Err(Error::ControlStore(format!(
                "the control database would take {bytes} bytes of RAM to hold off the stripe, \
                 over the {RAM_COPY_CEILING_BYTES}-byte ceiling; not suspending"
            )));
        }
        // The copy is installed BEFORE the file is closed, so no reader ever
        // meets a `None` database: the `ram` lock is taken after `live`'s
        // write lock, which every reader takes (shared) first — no reader is
        // inside the database at this point.
        *write_lock(&self.ram) = Some(copy.clone());
        if self.path.is_some() {
            // Dropping the database closes the file: redb has no explicit
            // close, and its `Drop` is what releases the descriptor and the
            // lock. Every handle upstream is an `Arc<ControlDb>`, so nothing
            // else holds the file.
            *live = None;
        }
        Ok(copy)
    }

    /// **Put the store back on the stripe.** Reopens the file (the mount
    /// must be back; a missing file is an error, never a fresh database),
    /// compares every document pointer on disk with the copy, and drops the
    /// copy. Refuses when not suspended.
    pub fn resume(&self) -> Result<ResumeReport> {
        let mut live = write_lock(&self.live);
        let Some(copy) = read_lock(&self.ram).clone() else {
            return Err(Error::ControlStore(
                "the control store is not suspended".to_owned(),
            ));
        };
        if live.is_none() {
            let path = self.path.as_ref().ok_or_else(|| {
                Error::ControlStore("no path to reopen the control store from".to_owned())
            })?;
            let db = redb::Database::open(path).map_err(|err| {
                faulted(
                    &format!("reopening the control store {}", path.display()),
                    err,
                )
            })?;
            *live = Some(db);
        }
        let db = live.as_ref().expect("installed above");
        let txn = db
            .begin_read()
            .map_err(|err| faulted("beginning the resume check", err))?;
        let table = txn
            .open_table(DOCUMENTS)
            .map_err(|err| faulted("opening the documents table", err))?;
        let mut matched = 0;
        let mut drift = Vec::new();
        for (name, ram_id) in &copy.docs {
            let disk = table
                .get(name.as_str())
                .map_err(|err| faulted("reading a document pointer", err))?
                .and_then(|v| BlobId::from_slice(v.value()));
            match disk {
                Some(id) if id == *ram_id => matched += 1,
                other => drift.push(format!(
                    "{name}: ram {ram_id} vs disk {}",
                    other
                        .map(|i| i.to_string())
                        .unwrap_or_else(|| "absent".to_owned())
                )),
            }
        }
        for entry in table
            .iter()
            .map_err(|err| faulted("scanning the documents table", err))?
        {
            let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
            if !copy.docs.contains_key(k.value()) {
                drift.push(format!(
                    "{}: ram absent vs disk {}",
                    k.value(),
                    BlobId::from_slice(v.value())
                        .map(|i| i.to_string())
                        .unwrap_or_else(|| "?".to_owned())
                ));
            }
        }
        drop(table);
        drop(txn);
        *write_lock(&self.ram) = None;
        Ok(ResumeReport {
            documents_matched: matched,
            drift,
            off_stripe_ms: crate::model::now_ms() - copy.taken_unix_ms,
        })
    }
}

/// The redb-backed [`ControlStore`]: one file, three tables.
///
/// Cheap to clone; every clone is the same database. redb serialises write
/// transactions internally, so two threads committing at once is a wait and
/// never a lock error — which is why [`ControlStore::commit_doc`] has one
/// failure mode (the expectation did not hold) where the reference store had
/// two (that, and *somebody else holds the lock*).
#[derive(Clone)]
pub struct RedbControlStore {
    db: Arc<ControlDb>,
}

impl std::fmt::Debug for RedbControlStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("RedbControlStore")
    }
}

fn faulted(what: &str, err: impl std::fmt::Display) -> Error {
    Error::ControlStore(format!("{what}: {err}"))
}

impl RedbControlStore {
    /// Open — creating if absent — the control database at `path`.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        let db = redb::Database::create(path).map_err(|err| {
            faulted(
                &format!("opening the control store {}", path.display()),
                err,
            )
        })?;
        Self::prepared(Some(path.to_path_buf()), db)
    }

    /// **The handle the runtime growth suspends and resumes.** The same
    /// `Arc` the secret store holds.
    pub fn control_db(&self) -> Arc<ControlDb> {
        Arc::clone(&self.db)
    }

    /// A control store with no file under it, for tests and for a `--dry-run`
    /// that must not touch the data path.
    pub fn in_memory() -> Result<Self> {
        let db = redb::Database::builder()
            .create_with_backend(redb::backends::InMemoryBackend::new())
            .map_err(|err| faulted("opening an in-memory control store", err))?;
        Self::prepared(None, db)
    }

    /// Create every table up front.
    ///
    /// redb refuses to *read* a table that has never been written, so a store
    /// whose tables were created lazily would answer a first read with a fault
    /// rather than with "empty" — the exact confusion
    /// [`ControlStore::read_doc`] exists to prevent.
    fn prepared(path: Option<std::path::PathBuf>, db: redb::Database) -> Result<Self> {
        let txn = db
            .begin_write()
            .map_err(|err| faulted("beginning the control store's first transaction", err))?;
        {
            txn.open_table(DOCUMENTS)
                .map_err(|err| faulted("creating the documents table", err))?;
            txn.open_table(BLOBS)
                .map_err(|err| faulted("creating the blobs table", err))?;
            txn.open_table(DOC_HISTORY)
                .map_err(|err| faulted("creating the document history table", err))?;
            txn.open_table(MARKS)
                .map_err(|err| faulted("creating the marks table", err))?;
        }
        txn.commit()
            .map_err(|err| faulted("committing the control store's first transaction", err))?;
        crate::secrets::RedbSecretStore::prepare(&db)?;
        Ok(RedbControlStore {
            db: Arc::new(ControlDb::new(path, db)),
        })
    }

    /// **The secret store in this same database** (2026-08-10).
    ///
    /// Handed out from here rather than opened separately, and that is the
    /// whole design: redb takes an exclusive file lock, so a second
    /// `redb::Database` over `control.redb` would fail. One file, one lock, one
    /// handle — and the secrets ride the transaction machinery the control
    /// plane already has instead of a second durability story beside it.
    ///
    /// It is **not** on [`ControlStore`]. That trait is about documents —
    /// `put_blob` / `commit_doc` / `history` — and its test decorators (the
    /// CAS-ignoring store `tests/control_cost.rs` proves the retry loop with,
    /// and the counting decorator) exist to break exactly those. A secret has
    /// none of that shape, so adding it to the trait would make every decorator
    /// carry a method it has no opinion about.
    pub fn secrets(&self) -> crate::secrets::RedbSecretStore {
        crate::secrets::RedbSecretStore::new(Arc::clone(&self.db))
    }

    /// The next sequence number for `name`, read inside the caller's write
    /// transaction so two committers cannot pick the same one.
    fn next_seq(
        table: &impl redb::ReadableTable<(&'static str, u64), &'static [u8]>,
        name: &str,
    ) -> Result<u64> {
        let mut range = table
            .range((name, 0u64)..=(name, u64::MAX))
            .map_err(|err| faulted("reading a document's history", err))?;
        match range.next_back() {
            None => Ok(1),
            Some(entry) => {
                let (key, _) = entry.map_err(|err| faulted("reading a history row", err))?;
                Ok(key.value().1 + 1)
            }
        }
    }
}

impl ControlStore for RedbControlStore {
    fn put_blob(&self, bytes: &[u8]) -> Result<BlobId> {
        let id = BlobId::of(bytes);
        self.db.write("a blob write", |txn| {
            {
                let mut table = txn
                    .open_table(BLOBS)
                    .map_err(|err| faulted("opening the blobs table", err))?;
                table
                    .insert(id.0.as_slice(), bytes)
                    .map_err(|err| faulted("storing a control blob", err))?;
            }
            Ok(Txn::Commit(id))
        })
    }

    fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
        let id = BlobId::of(bytes);
        self.db.write("a blob write", |txn| {
            {
                let mut table = txn
                    .open_table(BLOBS)
                    .map_err(|err| faulted("opening the blobs table", err))?;
                table
                    .insert(id.0.as_slice(), bytes)
                    .map_err(|err| faulted("storing a control blob", err))?;
            }
            Ok(Txn::CommitUnsynced(id))
        })
    }

    fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>> {
        self.db.read(
            "a blob read",
            |txn| {
                let table = txn
                    .open_table(BLOBS)
                    .map_err(|err| faulted("opening the blobs table", err))?;
                let found = table
                    .get(id.0.as_slice())
                    .map_err(|err| faulted("reading a control blob", err))?;
                Ok(found.map(|value| value.value().to_vec()))
            },
            |ram| Ok(ram.blobs.get(id).cloned()),
        )
    }

    fn read_doc(&self, name: &str) -> Result<Option<BlobId>> {
        self.db.read(
            "a document read",
            |txn| {
                let table = txn
                    .open_table(DOCUMENTS)
                    .map_err(|err| faulted("opening the documents table", err))?;
                let found = table
                    .get(name)
                    .map_err(|err| faulted("reading a control document pointer", err))?;
                match found {
                    None => Ok(None),
                    Some(value) => BlobId::from_slice(value.value())
                        .map(Some)
                        .ok_or_else(|| Error::CorruptDocument {
                            ref_name: name.to_owned(),
                            reason: format!(
                                "its stored pointer is {} bytes, not the {BLOB_ID_LEN} of a blob id",
                                value.value().len()
                            ),
                        }),
                }
            },
            |ram| Ok(ram.docs.get(name).copied()),
        )
    }

    fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool> {
        self.db.write("a control document commit", |txn| {
            {
                let mut documents = txn
                    .open_table(DOCUMENTS)
                    .map_err(|err| faulted("opening the documents table", err))?;
                let current = documents
                    .get(name)
                    .map_err(|err| faulted("reading a control document pointer", err))?
                    .and_then(|value| BlobId::from_slice(value.value()));
                if current != expected {
                    // Not an error: the caller re-reads and re-applies. Nothing
                    // was written yet, and the transaction is discarded.
                    return Ok(Txn::Discard(false));
                }
                documents
                    .insert(name, new.0.as_slice())
                    .map_err(|err| faulted("moving a control document pointer", err))?;

                // The control plane's audit trail, in the SAME transaction as
                // the pointer move. Two transactions could be interrupted
                // between them, and a history that can disagree with the
                // document it describes is worse than no history at all.
                let mut history = txn
                    .open_table(DOC_HISTORY)
                    .map_err(|err| faulted("opening the document history table", err))?;
                let seq = Self::next_seq(&history, name)?;
                let mut row = Vec::with_capacity(8 + BLOB_ID_LEN);
                row.extend_from_slice(&crate::model::now_ms().to_be_bytes());
                row.extend_from_slice(&new.0);
                history
                    .insert((name, seq), row.as_slice())
                    .map_err(|err| faulted("appending to the document history", err))?;
            }
            Ok(Txn::Commit(true))
        })
    }

    fn history(&self, name: &str) -> Result<Vec<DocVersion>> {
        self.db.read(
            "a history read",
            |txn| {
                let table = txn
                    .open_table(DOC_HISTORY)
                    .map_err(|err| faulted("opening the document history table", err))?;
                let range = table
                    .range((name, 0u64)..=(name, u64::MAX))
                    .map_err(|err| faulted("scanning a document's history", err))?;
                let mut out = Vec::new();
                for entry in range {
                    let (key, value) =
                        entry.map_err(|err| faulted("reading a history row", err))?;
                    let bytes = value.value();
                    if bytes.len() != 8 + BLOB_ID_LEN {
                        return Err(Error::CorruptDocument {
                            ref_name: name.to_owned(),
                            reason: format!(
                                "a history row is {} bytes, not {}",
                                bytes.len(),
                                8 + BLOB_ID_LEN
                            ),
                        });
                    }
                    let mut millis = [0u8; 8];
                    millis.copy_from_slice(&bytes[..8]);
                    out.push(DocVersion {
                        seq: key.value().1,
                        at: i64::from_be_bytes(millis),
                        blob: BlobId::from_slice(&bytes[8..]).expect("checked length"),
                    });
                }
                Ok(out)
            },
            // The trail is on the stripe and nothing wrote while it was off:
            // an audit read during the window is refused rather than answered
            // empty, which would read as "this document has no history".
            |ram| {
                Err(Error::OffStripe {
                    why: format!("{}; the audit trail is not held in RAM", ram.why),
                })
            },
        )
    }
}

/// A control store, ready to be handed to [`crate::AuthTable`] and
/// [`crate::RepoCatalog`].
pub type SharedControlStore = Arc<dyn ControlStore>;

/// The redb store at `path`, as a [`SharedControlStore`].
pub fn open_control_store(path: impl AsRef<Path>) -> Result<SharedControlStore> {
    Ok(Arc::new(RedbControlStore::open(path)?))
}

/// **The whole control plane at `path`: the documents and the secrets, over one
/// database handle.**
///
/// The pair, not two calls, because the two halves must share the `redb`
/// handle: opening the file twice is an exclusive-lock failure, and this is the
/// only place that fact has to be known. `ServerStores::open` is the caller.
pub fn open_control(
    path: impl AsRef<Path>,
) -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
    let (store, secrets, _) = open_control_with_db(path)?;
    Ok((store, secrets))
}

/// [`open_control`], and the [`ControlDb`] handle both halves share — what
/// the runtime data-set growth suspends and resumes (`gunnar-server`'s
/// `grow`). The handle is the same object either store reaches; it is
/// returned separately only because the store is type-erased on the way out.
pub fn open_control_with_db(
    path: impl AsRef<Path>,
) -> Result<(
    SharedControlStore,
    crate::secrets::RedbSecretStore,
    Arc<ControlDb>,
)> {
    let store = RedbControlStore::open(path)?;
    let secrets = store.secrets();
    let db = store.control_db();
    Ok((Arc::new(store), secrets, db))
}

/// An in-memory control store, as a [`SharedControlStore`].
pub fn memory_control_store() -> Result<SharedControlStore> {
    Ok(Arc::new(RedbControlStore::in_memory()?))
}

/// [`open_control`] with no file under it, for tests and for a `--dry-run` that
/// must not touch the data path.
pub fn memory_control() -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
    let store = RedbControlStore::in_memory()?;
    let secrets = store.secrets();
    Ok((Arc::new(store), secrets))
}

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

    #[test]
    fn a_blob_id_is_the_hash_of_the_bytes_and_round_trips_through_hex() {
        let id = BlobId::of(b"the identity table");
        assert_eq!(BlobId::from_hex(&id.to_string()), Some(id));
        assert_ne!(id, BlobId::of(b"the identity tabl3"));
        assert_eq!(BlobId::from_hex("not hex"), None);
        assert_eq!(BlobId::from_hex(""), None);
    }

    #[test]
    fn an_unwritten_document_reads_as_absent_and_a_written_one_reads_back() {
        let store = RedbControlStore::in_memory().expect("open");
        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), None);
        let id = store.put_blob(b"{}").expect("put");
        assert_eq!(
            store.get_blob(&id).expect("get").as_deref(),
            Some(&b"{}"[..])
        );
        assert!(store
            .commit_doc("refs/gunnar/auth", None, id)
            .expect("commit"));
        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
    }

    /// **T2 — the store off the stripe** (`DATA-SET-GROWTH-FLOW.md` G4):
    /// while suspended every read answers from the RAM copy byte for byte,
    /// every write is refused by name with `OffStripe`, the audit trail is
    /// refused rather than answered empty, and a resume finds no drift and
    /// takes writes again. Suspending twice and resuming an unsuspended
    /// store are both refusals.
    #[test]
    fn a_suspended_store_reads_from_ram_refuses_writes_and_resumes_without_drift() {
        use gunnar_store::SecretStore as _;
        let store = RedbControlStore::in_memory().expect("open");
        let secrets = store.secrets();
        let db = store.control_db();
        let id = store.put_blob(b"{\"principals\":1}").expect("put");
        assert!(store
            .commit_doc("refs/gunnar/auth", None, id)
            .expect("commit"));
        secrets
            .put("host-key", b"sealed bytes")
            .expect("put secret");
        assert!(!db.is_suspended());

        let copy = db.suspend("draining for a growth").expect("suspend");
        assert!(db.is_suspended());
        assert_eq!(copy.documents(), 1);
        assert!(copy.bytes() > 0);
        assert_eq!(copy.why, "draining for a growth");
        // Reads: served from the copy.
        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
        assert_eq!(
            store.get_blob(&id).expect("get").as_deref(),
            Some(&b"{\"principals\":1}"[..])
        );
        assert_eq!(store.read_doc("never-written").expect("read"), None);
        assert_eq!(
            secrets.get("host-key").expect("get").as_deref(),
            Some(&b"sealed bytes"[..])
        );
        assert_eq!(secrets.get("absent").expect("get"), None);
        assert_eq!(
            secrets
                .get_or_create("host-key", &mut || panic!("must not mint"))
                .expect("held"),
            b"sealed bytes"
        );
        // Writes: refused, by name, with the phase in the words.
        let refused = store.put_blob(b"new").unwrap_err();
        assert!(matches!(refused, Error::OffStripe { .. }), "{refused:?}");
        assert!(
            refused.to_string().contains("draining for a growth"),
            "{refused}"
        );
        assert!(refused.to_string().contains("retry"), "{refused}");
        let other = BlobId::of(b"other");
        assert!(matches!(
            store
                .commit_doc("refs/gunnar/auth", Some(id), other)
                .unwrap_err(),
            Error::OffStripe { .. }
        ));
        assert!(matches!(
            store.history("refs/gunnar/auth").unwrap_err(),
            Error::OffStripe { .. }
        ));
        let sec = secrets.put("host-key", b"x").unwrap_err().to_string();
        assert!(sec.contains("off the stripe"), "{sec}");
        let mint = secrets
            .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
            .unwrap_err()
            .to_string();
        assert!(mint.contains("will not be minted"), "{mint}");
        assert!(
            db.suspend("again").is_err(),
            "a second suspend is a refusal"
        );

        let report = db.resume().expect("resume");
        assert!(!db.is_suspended());
        assert_eq!(report.documents_matched, 1);
        assert!(report.drift.is_empty(), "{:?}", report.drift);
        assert!(
            db.resume().is_err(),
            "resuming an unsuspended store is a refusal"
        );
        // Back on the stripe: writes land, the trail is readable again.
        assert!(store
            .commit_doc("refs/gunnar/auth", Some(id), other)
            .expect("commit"));
        assert_eq!(store.history("refs/gunnar/auth").expect("history").len(), 2);
        assert_eq!(
            secrets
                .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
                .expect("mint"),
            b"minted"
        );
    }

    /// **The file really closes.** On the stripe the whole point of the
    /// suspend is that `umount /data` can succeed, which needs the descriptor
    /// gone: `/proc/self/fd` names the file before, not during, and again
    /// after. Linux only, like the mount it exists for.
    #[cfg(target_os = "linux")]
    #[test]
    fn a_suspended_file_store_holds_no_descriptor_on_its_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        let store = RedbControlStore::open(&path).expect("open");
        let db = store.control_db();
        let id = store.put_blob(b"doc").expect("put");
        assert!(store.commit_doc("d", None, id).expect("commit"));
        let holds = |p: &Path| -> bool {
            std::fs::read_dir("/proc/self/fd")
                .expect("procfs")
                .flatten()
                .filter_map(|e| std::fs::read_link(e.path()).ok())
                .any(|t| t == p)
        };
        assert!(holds(&path), "the open store holds its file");
        db.suspend("unmounting").expect("suspend");
        assert!(
            !holds(&path),
            "a suspended store holds NO descriptor on its file"
        );
        assert_eq!(store.read_doc("d").expect("read from ram"), Some(id));
        db.resume().expect("resume");
        assert!(holds(&path), "a resumed store holds its file again");
        assert_eq!(store.read_doc("d").expect("read from disk"), Some(id));
    }

    /// The compare-and-swap, seen to refuse. Without this the retry loop in
    /// `doc.rs` is decoration: a `commit_doc` that always returned `true` would
    /// satisfy every other test in this module.
    #[test]
    fn a_commit_against_the_wrong_expectation_is_refused_and_changes_nothing() {
        let store = RedbControlStore::in_memory().expect("open");
        let first = store.put_blob(b"one").expect("put");
        let second = store.put_blob(b"two").expect("put");
        assert!(store.commit_doc("doc", None, first).expect("commit"));

        // Somebody else already moved it: this writer's expectation is stale.
        assert!(
            !store.commit_doc("doc", None, second).expect("commit"),
            "a commit that expected an absent document was accepted against a written one"
        );
        assert_eq!(
            store.read_doc("doc").expect("read"),
            Some(first),
            "a refused commit still moved the pointer"
        );
        assert_eq!(
            store.history("doc").expect("history").len(),
            1,
            "a refused commit still appended to the audit trail"
        );

        // …and with the right expectation it lands.
        assert!(store
            .commit_doc("doc", Some(first), second)
            .expect("commit"));
        assert_eq!(store.read_doc("doc").expect("read"), Some(second));
    }

    /// The control plane's audit trail: every version, in order, and it is per-document.
    #[test]
    fn the_history_is_append_only_and_one_document_does_not_see_anothers() {
        let store = RedbControlStore::in_memory().expect("open");
        let mut previous = None;
        let mut expected = Vec::new();
        for text in [b"a".as_slice(), b"bb", b"ccc"] {
            let id = store.put_blob(text).expect("put");
            assert!(store.commit_doc("auth", previous, id).expect("commit"));
            previous = Some(id);
            expected.push(id);
        }
        let other = store.put_blob(b"catalog").expect("put");
        assert!(store.commit_doc("catalog", None, other).expect("commit"));

        let history = store.history("auth").expect("history");
        assert_eq!(
            history.iter().map(|v| v.blob).collect::<Vec<_>>(),
            expected,
            "the audit trail must list every version of the document, oldest first"
        );
        assert_eq!(
            history.iter().map(|v| v.seq).collect::<Vec<_>>(),
            vec![1, 2, 3]
        );
        assert_eq!(
            store.history("catalog").expect("history").len(),
            1,
            "one document's history picked up another document's versions"
        );
        assert_eq!(store.history("never-written").expect("history").len(), 0);

        // Every historical version is still readable, which is what makes this
        // an audit trail rather than a list of hashes.
        assert_eq!(
            store.get_blob(&history[0].blob).expect("get").as_deref(),
            Some(&b"a"[..])
        );
    }
}