gunnar-core 3.0.0

//! **Where the server's secrets live** — the SSH host key and the issuer
//! private key, in the control database beside the catalog and the identity
//! table.
//!
//! # Why they moved (the second half of the certificate-identity plan)
//!
//! On 2026-08-09 the catalog, the identity table and the issuer *certificate*
//! moved out of `control.git` and into `control.redb`. The secrets did not, and
//! `gunnar-server`'s `stores.rs` said why in as many words: it would have been
//! *"a second migration riding on the first"*. It also named the price it was
//! paying to defer them:
//!
//! > *"It does mean the control **repository** is still opened, and therefore
//! > still needs an installed storage engine — so it is opened **lazily**, on
//! > the first call to `ServerStores::secrets`."*
//!
//! That lazy open is gone, and with it the last reason a `gunnar-server`
//! process touches a storage engine in order to read its own host key. The
//! sentence *"the catalog and the identity table are reachable in a process
//! that carries no engine"* was true; **"the server is"** now is too.
//!
//! # What this module is NOT
//!
//! It is not crypto. [`crate::seal`] wraps the *value* — `SealedSecretStore` is
//! a decorator over [`SecretStore`], the KEK is a file beside the database, and
//! what reaches this module is already ciphertext. This module decides only
//! **where sealed bytes are kept**. Sealed bytes in, sealed bytes out.
//!
//! # The table, and the one byte in front of every value
//!
//! ```text
//!   secrets      key → tag ‖ value      tag 0x01 = a value, 0x00 = a tombstone
//! ```
//!
//! A tombstone rather than a removed row, because of the migration below: the
//! legacy Arrow log is only consulted when redb has **no row at all** for a
//! key, so a `delete` that removed the row would let the next `get` fall
//! through and resurrect the deleted secret out of the old file. A row that
//! says *"deleted"* cannot do that. It also happens to be the honest record —
//! the deletion is a fact, and one byte is what it costs to keep it.
//!
//! # What redb gives that the Arrow log had to build
//!
//! [`SecretStore::get_or_create`] must be atomic — *"the host key must never be
//! regenerated by a race"* — and the Arrow log achieved that by taking its
//! writer mutex **before** the existence check. Here the existence check and
//! the insert are the same redb write transaction, and redb serialises write
//! transactions internally. The atomicity is the store's, not a mutex this
//! module holds.
//!
//! # 🔴 What redb does NOT give, recorded as a fact rather than glossed
//!
//! `arrow_log.rs` compacts a *superseded* secret out of its file immediately,
//! and its comment says why: *"a rotated SSH host key must actually stop
//! existing"*. redb is a copy-on-write B-tree. An `insert` over an existing key
//! writes a new page and frees the old one; the freed page is reused by a later
//! write, but until it is, **the previous ciphertext is still bytes in
//! `control.redb`**.
//!
//! Two things bound that, and neither is an accident:
//!
//! * what is in those pages is **ciphertext**, because everything on the boot
//!   path goes through [`crate::SealedSecretStore`] and the KEK is a separate
//!   file (`secrets.kek`) that a database copy does not carry; and
//! * it is bounded by page reuse rather than by time — it is residue, not an
//!   archive: nothing indexes it and no read path can reach it.
//!
//! It is still a narrowing against the Arrow log's explicit fold, and a
//! `gunnar control gc` that called `redb::Database::compact` is where it would
//! be closed. **Not built here.** Named.
//!
//! # One process, one lock
//!
//! redb takes an exclusive file lock on `control.redb`, so a second gunnar
//! process on the same data root cannot open it. That was already true of the
//! catalog and the identity table, and it is what this change makes true of the
//! host key as well. The bound does not get *tighter*: the Arrow log took an
//! `O_EXCL` lock of its own on the same data root, so two servers already could
//! not share it. What changes is that the two locks become one.

use std::sync::Arc;

use gunnar_store::{Result as StoreResult, SecretStore};
use redb::ReadableTable;

use crate::error::Result;

/// Every secret this server keeps, one row each.
const SECRETS: redb::TableDefinition<'static, &str, &[u8]> = redb::TableDefinition::new("secrets");

/// A row that carries a value.
const TAG_VALUE: u8 = 0x01;

/// A row that records a deletion. See the module header.
const TAG_TOMBSTONE: u8 = 0x00;

/// Where the pre-2026-08-10 secret log lives, relative to the control
/// directory, newest layout first.
///
/// Two entries because the layout moved once and an appliance may predate the
/// move: `5f0fa37` (the `gunnar-server` split) put it at `gunnar-secrets/`, and
/// the `FsStore` that came before that put it at `gunnar-store/`. Probing both
/// costs two `Path::is_dir` calls on the boot path and is the difference
/// between an old appliance keeping its host key and an old appliance minting a
/// new one.
pub const LEGACY_SECRET_DIRS: &[&str] = &["gunnar-secrets", "gunnar-store"];

/// The file `ArrowLogStore` keeps its secrets in. Named here so the probe below
/// checks for the log itself rather than for a directory that happens to exist.
const LEGACY_SECRET_FILE: &str = "secrets.arrow";

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

/// Decode a stored row: `Some(bytes)` for a value, `None` for a tombstone.
///
/// A row that is neither is a fault and not an absence — a secret store that
/// answered "no host key" because it could not parse its own row would mint a
/// second host key, which is the one outcome this whole module exists to
/// prevent.
fn decode(key: &str, row: &[u8]) -> StoreResult<Option<Vec<u8>>> {
    match row.split_first() {
        Some((&TAG_VALUE, rest)) => Ok(Some(rest.to_vec())),
        Some((&TAG_TOMBSTONE, _)) => Ok(None),
        _ => Err(gunnar_store::Error::backend(format!(
            "the control database's row for the secret {key} is {} bytes and does not begin with \
             a known tag; refusing to read it as absent, because the caller's next move would be \
             to mint a replacement",
            row.len()
        ))),
    }
}

fn encode(value: &[u8]) -> Vec<u8> {
    let mut row = Vec::with_capacity(1 + value.len());
    row.push(TAG_VALUE);
    row.extend_from_slice(value);
    row
}

/// **The secret store in the control database.**
///
/// Cheap to clone; every clone is the same database, and the same one
/// [`crate::RedbControlStore`] holds. It is deliberately not a second
/// `redb::Database` over the same file — redb takes an exclusive lock, so a
/// second open in the same process would fail, and the shared handle is what
/// makes "one file, one lock" true.
#[derive(Clone)]
pub struct RedbSecretStore {
    db: Arc<crate::control::ControlDb>,
}

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

impl RedbSecretStore {
    /// Build one over an already-open control database.
    ///
    /// `pub(crate)` on purpose: the only supported way to get one is from the
    /// control store that owns the handle, so there is no path by which a
    /// caller opens a second database over the same file.
    pub(crate) fn new(db: Arc<crate::control::ControlDb>) -> Self {
        RedbSecretStore { db }
    }

    /// Create the table if it has never been written.
    ///
    /// Same reason [`crate::RedbControlStore::prepared`] does it for its three:
    /// redb refuses to *read* a table that does not exist, so a first `get`
    /// would be a fault rather than an absence — and an absence is exactly what
    /// a first boot has to be able to say.
    pub(crate) fn prepare(db: &redb::Database) -> Result<()> {
        let txn = db.begin_write().map_err(|err| {
            crate::Error::ControlStore(format!("beginning the secrets table's creation: {err}"))
        })?;
        txn.open_table(SECRETS).map_err(|err| {
            crate::Error::ControlStore(format!("creating the secrets table: {err}"))
        })?;
        txn.commit().map_err(|err| {
            crate::Error::ControlStore(format!("committing the secrets table's creation: {err}"))
        })?;
        Ok(())
    }

    /// A raw row decoded: `Some(value)` or `None` for a tombstone.
    pub(crate) fn decode_row_of(key: &str, row: &[u8]) -> StoreResult<Option<Vec<u8>>> {
        decode(key, row)
    }

    /// Every row of the secrets table, raw (tag ‖ sealed bytes) — what
    /// `ControlDb::suspend` copies into RAM. Sealed at rest, sealed in RAM.
    pub(crate) fn rows(
        txn: &redb::ReadTransaction,
    ) -> Result<std::collections::BTreeMap<String, Vec<u8>>> {
        let table = txn.open_table(SECRETS).map_err(|err| {
            crate::Error::ControlStore(format!("opening the secrets table: {err}"))
        })?;
        let mut out = std::collections::BTreeMap::new();
        for entry in table.iter().map_err(|err| {
            crate::Error::ControlStore(format!("scanning the secrets table: {err}"))
        })? {
            let (k, v) = entry.map_err(|err| {
                crate::Error::ControlStore(format!("reading a secret row: {err}"))
            })?;
            out.insert(k.value().to_owned(), v.value().to_vec());
        }
        Ok(out)
    }

    /// Write a row, value or tombstone, in one transaction. Refused while the
    /// store is off the stripe (`Error::OffStripe`, in the backend error's
    /// words): nothing may be written to a file that is closed, and a secret
    /// minted into RAM alone would be a secret the next boot does not have.
    fn write(&self, key: &str, row: &[u8]) -> StoreResult<()> {
        self.db
            .write("a secret write", |txn| {
                {
                    let mut table = txn
                        .open_table(SECRETS)
                        .map_err(|err| faulted("opening the secrets table", err))?;
                    table
                        .insert(key, row)
                        .map_err(|err| faulted(&format!("storing the secret {key}"), err))?;
                }
                Ok(crate::control::Txn::Commit(()))
            })
            .map_err(off_stripe)
    }

    /// Three-state read: a value, a tombstone, or no row at all.
    ///
    /// The third state is what the migration below needs and what
    /// [`SecretStore::get`] cannot express — see [`MigratingSecretStore`].
    fn row(&self, key: &str) -> StoreResult<Option<Option<Vec<u8>>>> {
        self.db
            .read(
                "a secret read",
                |txn| {
                    let table = txn
                        .open_table(SECRETS)
                        .map_err(|err| faulted("opening the secrets table", err))?;
                    let found = table
                        .get(key)
                        .map_err(|err| faulted(&format!("reading the secret {key}"), err))?;
                    match found {
                        None => Ok(None),
                        Some(value) => decode(key, value.value()).map(Some).map_err(Into::into),
                    }
                },
                |ram| match ram.secret_row(key) {
                    None => Ok(None),
                    Some(row) => decode(key, row).map(Some).map_err(Into::into),
                },
            )
            .map_err(off_stripe)
    }
}

/// A control-store error as the secret store's callers see it: the
/// `OffStripe` refusal keeps its own words, everything else is a backend
/// fault.
fn off_stripe(err: crate::Error) -> gunnar_store::Error {
    match err {
        crate::Error::Store(inner) => inner,
        other => gunnar_store::Error::backend(other.to_string()),
    }
}

impl SecretStore for RedbSecretStore {
    fn get(&self, key: &str) -> StoreResult<Option<Vec<u8>>> {
        Ok(self.row(key)?.flatten())
    }

    fn put(&self, key: &str, value: &[u8]) -> StoreResult<()> {
        self.write(key, &encode(value))
    }

    fn delete(&self, key: &str) -> StoreResult<()> {
        self.write(key, &[TAG_TOMBSTONE])
    }

    fn get_or_create(
        &self,
        key: &str,
        mk: &mut dyn FnMut() -> StoreResult<Vec<u8>>,
    ) -> StoreResult<Vec<u8>> {
        // Off the stripe: a value the copy holds is the answer; an absent one
        // is a refusal, never a mint — a key minted into RAM would be a key
        // the file never gets, which for the host key is the "REMOTE HOST
        // IDENTIFICATION HAS CHANGED" this whole store exists to prevent.
        if self.db.is_suspended() {
            if let Some(bytes) = self.get(key)? {
                return Ok(bytes);
            }
            return Err(off_stripe(crate::Error::OffStripe {
                why: format!(
                    "{}; the secret {key} is not in the RAM copy and will not be minted off the stripe",
                    self.db.ram_copy().map(|r| r.why).unwrap_or_default()
                ),
            }));
        }
        // The existence check and the insert are ONE write transaction, and
        // redb serialises write transactions. The loser of a race blocks in
        // `begin_write` and then finds the winner's value, so `mk` runs once
        // per absent key and never twice.
        //
        // `mk` runs while the write lock is held. That is deliberate and it is
        // what makes the guarantee hold; it costs the microseconds it takes to
        // mint an ed25519 key, on a path that runs at most once per key per
        // installation.
        self.db
            .write("a secret get-or-create", |txn| {
                let created = {
                    let mut table = txn
                        .open_table(SECRETS)
                        .map_err(|err| faulted("opening the secrets table", err))?;
                    let existing = table
                        .get(key)
                        .map_err(|err| faulted(&format!("reading the secret {key}"), err))?
                        .map(|value| decode(key, value.value()))
                        .transpose()?
                        .flatten();
                    match existing {
                        Some(bytes) => {
                            // Nothing to commit. Discarding the transaction is
                            // cheaper than committing an empty one and cannot
                            // fail.
                            drop(table);
                            return Ok(crate::control::Txn::Discard(bytes));
                        }
                        None => {
                            // A failing `mk` drops the transaction, so a mint
                            // that could not produce bytes leaves the key
                            // absent rather than half-written.
                            let value = mk()?;
                            table
                                .insert(key, encode(&value).as_slice())
                                .map_err(|err| {
                                    faulted(&format!("storing the secret {key}"), err)
                                })?;
                            value
                        }
                    }
                };
                Ok(crate::control::Txn::Commit(created))
            })
            .map_err(off_stripe)
    }
}

/// **Read the old place, write the new one.** The migration, and it is not a
/// flag day.
///
/// ```text
///   get(k)             redb row?  ── value ──▶ it
///                          │
///                       tombstone ──▶ None            (never falls through)
///                          │
///                       no row    ──▶ legacy Arrow log ──▶ adopt into redb
///
///   get_or_create(k)   redb row?  ── value ──▶ it
///                          │
///                       no row    ──▶ legacy? ── yes ──▶ adopt it, return it
///                                        └──── no ───▶ mk(), store, return
/// ```
///
/// # Why this is the shape, and what it costs to get it wrong
///
/// `serve.rs` calls `ensure_host_key` on **every boot**. The host key is the
/// one secret whose whole purpose is to be the same bytes it was yesterday: a
/// server that mints a fresh one gives every client
/// `REMOTE HOST IDENTIFICATION HAS CHANGED`, which is indistinguishable from an
/// attack. So the adoption of a legacy key happens **inside** the same
/// `get_or_create` that would otherwise mint, not in a separate migration step
/// that could be skipped, reordered or half-run:
///
/// * an appliance with a host key in the Arrow log boots, finds no redb row,
///   reads the Arrow log, writes those exact bytes into redb and returns them —
///   **same key, same fingerprint, no warning**;
/// * the next boot finds the redb row and never opens the Arrow log at all;
/// * a fresh installation has neither, mints once, and never grows an Arrow
///   log in the first place;
/// * a boot that is interrupted between reading the old value and committing
///   the new one has changed nothing, and the next boot repeats the adoption.
///
/// **The Arrow log is never written and never deleted by this type.** It stays
/// on disk, byte for byte, as the rollback: a binary from before this change,
/// pointed at the same data root, still finds its host key exactly where it
/// left it. Removing it is a later, separate decision.
///
/// # The legacy handle is `Option`, and absent is the normal case
///
/// It is `None` when there is no `secrets.arrow` under the control directory,
/// which is every installation created after this change. A `None` legacy is
/// not a degraded [`MigratingSecretStore`]; it is the steady state, and the
/// type collapses to [`RedbSecretStore`] with one extra `Option` check.
pub struct MigratingSecretStore {
    new: RedbSecretStore,
    legacy: Option<Arc<dyn SecretStore>>,
}

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

impl MigratingSecretStore {
    /// Wrap `new`, falling back to `legacy` for keys redb has never seen.
    pub fn new(new: RedbSecretStore, legacy: Option<Arc<dyn SecretStore>>) -> Self {
        MigratingSecretStore { new, legacy }
    }

    /// The legacy value for `key`, or `None` if there is no legacy store or it
    /// does not hold one.
    fn legacy_get(&self, key: &str) -> StoreResult<Option<Vec<u8>>> {
        match &self.legacy {
            None => Ok(None),
            Some(store) => store.get(key),
        }
    }
}

impl SecretStore for MigratingSecretStore {
    fn get(&self, key: &str) -> StoreResult<Option<Vec<u8>>> {
        // A tombstone answers `None` here without consulting the legacy store:
        // `row` distinguishes "deleted" from "never seen", and only the second
        // one falls through.
        if let Some(answer) = self.new.row(key)? {
            return Ok(answer);
        }
        let Some(bytes) = self.legacy_get(key)? else {
            return Ok(None);
        };
        // Adopt on read, so a plain `get` also finishes the migration. `put` is
        // safe to race: two threads adopting the same legacy value write the
        // same bytes.
        tracing::info!(
            key,
            "adopting a secret out of the legacy Arrow log into the control database; the old \
             log is left on disk untouched"
        );
        self.new.put(key, &bytes)?;
        Ok(Some(bytes))
    }

    fn put(&self, key: &str, value: &[u8]) -> StoreResult<()> {
        // New writes go to the new place only. The legacy log is never written
        // again — a value in two places is a value that can disagree.
        self.new.put(key, value)
    }

    fn delete(&self, key: &str) -> StoreResult<()> {
        // A tombstone, not a removal: see the module header. This is why a
        // delete does not resurrect out of the Arrow log on the next read.
        self.new.delete(key)
    }

    fn get_or_create(
        &self,
        key: &str,
        mk: &mut dyn FnMut() -> StoreResult<Vec<u8>>,
    ) -> StoreResult<Vec<u8>> {
        // **The whole migration, in one atomic step.** The legacy read is the
        // first thing the mint closure tries, so it runs inside the same redb
        // write transaction that decides whether to mint at all. There is no
        // window in which one thread reads the legacy key and another mints a
        // replacement.
        let mut adopted = false;
        let value = self.new.get_or_create(key, &mut || {
            if let Some(bytes) = self.legacy_get(key)? {
                adopted = true;
                return Ok(bytes);
            }
            mk()
        })?;
        if adopted {
            tracing::info!(
                key,
                "carried a secret forward from the legacy Arrow log into the control database \
                 rather than minting a new one; the old log is left on disk untouched"
            );
        }
        Ok(value)
    }
}

/// **The legacy Arrow secret log under `control_dir`, if there is one.**
///
/// `Ok(None)` when no `secrets.arrow` exists under any known layout — the
/// normal answer on every installation made after 2026-08-10, and the reason a
/// fresh appliance never creates an Arrow log again.
///
/// It opens what is there and **creates nothing**. That is the difference
/// between this and the `open_secret_log` it replaces, which did
/// `create_dir_all` on the way past and therefore minted an empty Arrow log
/// beside every repository the server ever opened.
///
/// No storage engine is involved: `ArrowLogStore` is a plain reader of a plain
/// file in `gunnar-store`, and the engine was only ever in the path because the
/// log was reached through `RepoStore::secrets()` on a repository the engine
/// had opened. It never needed to be.
pub fn legacy_secret_log(
    control_dir: &std::path::Path,
    hash_kind: gix_hash::Kind,
) -> Result<Option<Arc<dyn SecretStore>>> {
    let Some(dir) = LEGACY_SECRET_DIRS
        .iter()
        .map(|name| control_dir.join(name))
        .find(|dir| dir.join(LEGACY_SECRET_FILE).is_file())
    else {
        return Ok(None);
    };
    let log = crate::arrow_log::ArrowLogStore::open(&dir, hash_kind).map_err(|err| {
        crate::Error::ControlStore(format!(
            "opening the legacy secret log at {} to carry its secrets into the control database: \
             {err}. It is read here and never written; if a previous gunnar was killed, its \
             O_EXCL lock may still be there",
            dir.display()
        ))
    })?;
    tracing::info!(
        dir = %dir.display(),
        "a legacy Arrow secret log is present; secrets not yet in the control database will be \
         read from it once and carried forward"
    );
    Ok(Some(Arc::new(log.secrets()) as Arc<dyn SecretStore>))
}

#[cfg(test)]
mod tests {
    use super::*;
    use gunnar_store::mem::MemSecretStore;
    use gunnar_store::SecretStoreExt;

    fn store() -> RedbSecretStore {
        let db = redb::Database::builder()
            .create_with_backend(redb::backends::InMemoryBackend::new())
            .expect("in-memory control database");
        RedbSecretStore::prepare(&db).expect("prepare");
        RedbSecretStore::new(Arc::new(crate::control::ControlDb::in_memory_for_tests(db)))
    }

    #[test]
    fn a_secret_round_trips_and_an_unwritten_one_is_absent() {
        let store = store();
        assert_eq!(store.get("ssh/host_ed25519").expect("get"), None);
        store.put("ssh/host_ed25519", b"key").expect("put");
        assert_eq!(
            store.get("ssh/host_ed25519").expect("get").as_deref(),
            Some(&b"key"[..])
        );
    }

    /// A deleted secret reads as absent, and the *value* is gone from the row
    /// rather than merely unreferenced.
    #[test]
    fn a_deleted_secret_is_absent_and_its_bytes_are_not_the_row() {
        let store = store();
        store.put("k", b"secret-bytes").expect("put");
        store.delete("k").expect("delete");
        assert_eq!(store.get("k").expect("get"), None);
        // The row is a bare tombstone, so the value is not what a reader of the
        // table would find.
        assert_eq!(store.row("k").expect("row"), Some(None));
        // Deleting what is not there is not an error.
        store.delete("never-written").expect("delete absent");
    }

    #[test]
    fn get_or_create_mints_once_and_then_returns_the_same_bytes() {
        let store = store();
        let mut calls = 0usize;
        let first = store
            .get_or_create("k", &mut || {
                calls += 1;
                Ok(b"minted".to_vec())
            })
            .expect("first");
        let second = store
            .get_or_create("k", &mut || {
                calls += 1;
                Ok(b"different".to_vec())
            })
            .expect("second");
        assert_eq!(first, b"minted".to_vec());
        assert_eq!(second, first, "a second call minted a new value");
        assert_eq!(calls, 1);
    }

    /// A mint that fails must leave the key absent, not half-written.
    #[test]
    fn a_failed_mint_stores_nothing() {
        let store = store();
        let err = store
            .get_or_create("k", &mut || {
                Err(gunnar_store::Error::backend("the entropy source is gone"))
            })
            .expect_err("the mint failed");
        assert!(format!("{err}").contains("entropy"), "{err}");
        assert_eq!(store.get("k").expect("get"), None);
    }

    /// **The host-key migration, in the shape `serve.rs` runs it.**
    ///
    /// The red half is the whole point: the same call against a store with
    /// *no* legacy behind it must mint different bytes. Without it, "the key
    /// survived" is a claim a store that always returned the legacy value would
    /// also satisfy.
    #[test]
    fn a_legacy_host_key_is_carried_forward_rather_than_reminted() {
        let legacy = Arc::new(MemSecretStore::new());
        legacy
            .put("ssh/host_ed25519", b"the-live-host-key")
            .expect("seed");

        let new = store();
        let migrating = MigratingSecretStore::new(new.clone(), Some(legacy.clone()));
        let got = migrating
            .get_or_create_with("ssh/host_ed25519", || Ok(b"A-BRAND-NEW-KEY".to_vec()))
            .expect("first boot after the migration");
        assert_eq!(
            got,
            b"the-live-host-key".to_vec(),
            "the appliance minted a new host key instead of keeping the one it had"
        );

        // It landed in redb, so the second boot never touches the Arrow log.
        assert_eq!(
            new.get("ssh/host_ed25519").expect("get").as_deref(),
            Some(&b"the-live-host-key"[..])
        );
        let without_legacy = MigratingSecretStore::new(new.clone(), None);
        assert_eq!(
            without_legacy
                .get_or_create_with("ssh/host_ed25519", || Ok(b"A-BRAND-NEW-KEY".to_vec()))
                .expect("second boot"),
            b"the-live-host-key".to_vec()
        );

        // RED HALF: with nothing behind it and nothing in redb, the same call
        // mints. If this passed too, the assertion above would be measuring
        // nothing.
        let fresh = MigratingSecretStore::new(store(), None);
        assert_eq!(
            fresh
                .get_or_create_with("ssh/host_ed25519", || Ok(b"A-BRAND-NEW-KEY".to_vec()))
                .expect("fresh install"),
            b"A-BRAND-NEW-KEY".to_vec()
        );
    }

    /// A plain `get` finishes the migration too, and the legacy log is not
    /// written on the way past.
    #[test]
    fn a_read_adopts_and_does_not_write_the_legacy_log() {
        let legacy = Arc::new(MemSecretStore::new());
        legacy.put("pki/issuer_key", b"issuer").expect("seed");
        let new = store();
        let migrating = MigratingSecretStore::new(new.clone(), Some(legacy.clone()));

        assert_eq!(
            migrating.get("pki/issuer_key").expect("get").as_deref(),
            Some(&b"issuer"[..])
        );
        assert_eq!(
            new.get("pki/issuer_key").expect("get").as_deref(),
            Some(&b"issuer"[..]),
            "the read did not adopt the value into the control database"
        );

        // A later `put` goes to the new store only; the legacy log still holds
        // what it held, which is what makes a rollback possible.
        migrating.put("pki/issuer_key", b"rotated").expect("put");
        assert_eq!(
            legacy.get("pki/issuer_key").expect("get").as_deref(),
            Some(&b"issuer"[..]),
            "the legacy log was written; it must be left byte for byte as it was"
        );
        assert_eq!(
            migrating.get("pki/issuer_key").expect("get").as_deref(),
            Some(&b"rotated"[..])
        );
    }

    /// **The resurrection bug the tombstone exists to prevent.**
    ///
    /// Delete a key that is still in the legacy log, then read it. Without a
    /// tombstone the read falls through to the Arrow log and hands back the
    /// secret that was just deleted.
    #[test]
    fn a_deleted_secret_does_not_come_back_out_of_the_legacy_log() {
        let legacy = Arc::new(MemSecretStore::new());
        legacy.put("k", b"old").expect("seed");
        let migrating = MigratingSecretStore::new(store(), Some(legacy));
        assert_eq!(
            migrating.get("k").expect("get").as_deref(),
            Some(&b"old"[..])
        );
        migrating.delete("k").expect("delete");
        assert_eq!(
            migrating.get("k").expect("get"),
            None,
            "a deleted secret was resurrected out of the legacy Arrow log"
        );
    }

    /// Two callers racing on an absent key agree on one value, which is the
    /// property `ensure_host_key` rests on.
    ///
    /// `std::thread::scope`, no pool, no rayon — and this is a race probe, not
    /// a workload.
    #[test]
    fn get_or_create_is_atomic_across_threads() {
        let store = store();
        let mints = std::sync::atomic::AtomicUsize::new(0);
        let values: Vec<Vec<u8>> = std::thread::scope(|scope| {
            let handles: Vec<_> = (0..8)
                .map(|_| {
                    scope.spawn(|| {
                        store
                            .get_or_create("ssh/host_ed25519", &mut || {
                                let n = mints.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                                Ok(format!("key-{n}").into_bytes())
                            })
                            .expect("get_or_create")
                    })
                })
                .collect();
            handles
                .into_iter()
                .map(|h| h.join().expect("join"))
                .collect()
        });
        assert_eq!(
            mints.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the host key was minted more than once by a race"
        );
        assert!(
            values.windows(2).all(|w| w[0] == w[1]),
            "racing callers were handed different host keys"
        );
    }

    /// A row with an unknown tag is a fault, never an absence. An absence would
    /// make the caller mint a replacement host key.
    #[test]
    fn an_unreadable_row_is_a_fault_and_not_an_absence() {
        let err = decode("ssh/host_ed25519", &[0x7f, 1, 2, 3]).expect_err("unknown tag");
        assert!(format!("{err}").contains("known tag"), "{err}");
        assert!(
            decode("k", &[]).is_err(),
            "an empty row must not read as absent"
        );
    }
}