gunnar_core/

txlog_tee.rs

//! **The tee: every durable mutation goes through the transaction log
//! BEFORE it reaches the store.** `.nornir/twin-transaction-log.md` §"Before,
//! not after".
//!
//! Three decorators, one per store the primary mutates: [`LoggedGit`] around
//! the engine's `GitServe`, [`LoggedControlStore`] around the control store,
//! [`LoggedSecretStore`] around the secret store. (The LFS tee lives in the
//! server, which is the crate that sees `gunnar-lfs`.) Each one appends the
//! frame — durable — and only then calls the store. A log that cannot be
//! written is an error the caller sees, so a primary never mutates what its
//! twin will never hear of.
//!
//! Reads pass straight through. Nothing here changes a signature; the server
//! wraps at the one place each store is constructed.
//!
//! # CAS is recorded, not resolved
//!
//! `put_refs_cas`, `update_ref` and `commit_doc` carry the claim the caller
//! made (`old` / `expected`). The frame carries the same claim; the twin makes
//! it against its own copy and gets the same answer the primary got, because
//! it applies exactly the primary's sequence from the same state. A lost CAS
//! on the primary is therefore a frame the twin replays to the same `false`.

use std::sync::{Arc, Mutex};

use gunnar_store::git_storage_trait::{
    Caps, Extent, GcReport, GitOps, GitServe, ObjType, Oid, PackProgress, PackStats, ReachSet,
    RefCas, RefRow, RefUpdate, Stored, TxId,
};

use crate::control::{BlobId, ControlStore, DocVersion};
use crate::txlog::{Kind, Log, Meta, RefEdit};

fn hex(bytes: &[u8]) -> String {
    hex::encode(bytes)
}

fn blob_hex(id: &BlobId) -> String {
    hex(id.as_bytes())
}

// ── git ─────────────────────────────────────────────────────────────────────

/// A `GitServe` whose every write is a frame first.
pub struct LoggedGit {
    inner: Arc<dyn GitServe>,
    log: Arc<Log>,
    repo: String,
    hash: String,
}

impl LoggedGit {
    /// Wrap `inner` for repository `repo` (`namespace/name`) with hash kind
    /// `hash` (`sha1` / `sha256`).
    pub fn new(inner: Arc<dyn GitServe>, log: Arc<Log>, repo: impl Into<String>, hash: impl Into<String>) -> Self {
        Self {
            inner,
            log,
            repo: repo.into(),
            hash: hash.into(),
        }
    }

    /// The wrapped store.
    pub fn inner(&self) -> &Arc<dyn GitServe> {
        &self.inner
    }

    fn log_pack(&self, bytes: &[u8]) -> anyhow::Result<u64> {
        let meta = Meta::Pack {
            repo: self.repo.clone(),
            hash: self.hash.clone(),
        };
        Ok(self.log.append(Kind::Pack, &meta, bytes)?)
    }

    fn log_refs(&self, atomic: bool, edits: Vec<RefEdit>) -> anyhow::Result<u64> {
        let meta = Meta::Refs {
            repo: self.repo.clone(),
            atomic,
            edits,
        };
        Ok(self.log.append(Kind::Refs, &meta, &[])?)
    }
}

/// A plain `put_refs` update as the log records it: no claim about the old
/// value.
pub fn edit_of_update(u: &RefUpdate) -> RefEdit {
    RefEdit {
        name: u.name.clone(),
        old: None,
        new: u.target.clone(),
        peeled: u.peeled.clone(),
        symref: u.symref_target.clone(),
    }
}

/// A CAS edit as the log records it: the claim travels with it.
pub fn edit_of_cas(c: &RefCas<'_>) -> RefEdit {
    RefEdit {
        name: c.name.clone(),
        old: Some(c.old.map(hex)),
        new: c.new.map(hex),
        peeled: c.peeled.as_deref().map(hex),
        symref: None,
    }
}

impl GitOps for LoggedGit {
    fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_pack(pack)?;
        let _m = self.log.mutation();
        self.log_refs(false, refs.iter().map(edit_of_update).collect())?;
        self.inner.put(pack, refs)
    }
    fn put_pack(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_pack(bytes)?;
        self.inner.put_pack(bytes)
    }
    fn put_pack_observed(&self, bytes: &[u8], progress: &dyn PackProgress) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_pack(bytes)?;
        self.inner.put_pack_observed(bytes, progress)
    }
    fn publishes_pack_progress(&self) -> bool {
        self.inner.publishes_pack_progress()
    }
    fn put_refs(&self, updates: &[RefUpdate]) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_refs(false, updates.iter().map(edit_of_update).collect())?;
        self.inner.put_refs(updates)
    }
    fn get(&self, oid: Oid<'_>) -> anyhow::Result<Option<Stored>> {
        self.inner.get(oid)
    }
    fn has(&self, oid: Oid<'_>) -> anyhow::Result<bool> {
        self.inner.has(oid)
    }
    fn size(&self, oid: Oid<'_>) -> anyhow::Result<Option<u64>> {
        self.inner.size(oid)
    }
    fn extents(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<Extent>>> {
        self.inner.extents(oids)
    }
    fn refs(&self) -> anyhow::Result<Vec<RefRow>> {
        self.inner.refs()
    }
    fn update_ref(&self, name: &str, old: Option<Oid<'_>>, new: Option<Oid<'_>>) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_refs(
            true,
            vec![RefEdit {
                name: name.to_string(),
                old: Some(old.map(hex)),
                new: new.map(hex),
                peeled: None,
                symref: None,
            }],
        )?;
        self.inner.update_ref(name, old, new)
    }
    fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_refs(true, edits.iter().map(edit_of_cas).collect())?;
        self.inner.put_refs_cas(edits)
    }
    fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Vec<Vec<u8>>> {
        self.inner.reachable(want, have)
    }
    fn gc(&self) -> anyhow::Result<GcReport> {
        // A repack is not a transaction: the twin grows its own packs from the
        // ones as pushed and may gc on its own clock.
        self.inner.gc()
    }
}

impl GitServe for LoggedGit {
    fn read(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, Vec<u8>)>> {
        self.inner.read(oid)
    }
    fn header(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, u64)>> {
        self.inner.header(oid)
    }
    fn sizes(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<u64>>> {
        self.inner.sizes(oids)
    }
    fn index_now(&self) -> anyhow::Result<usize> {
        self.inner.index_now()
    }
    fn head(&self) -> anyhow::Result<Option<RefRow>> {
        self.inner.head()
    }
    fn set_head(&self, target: &str) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_refs(
            false,
            vec![RefEdit {
                name: "HEAD".to_string(),
                old: None,
                new: None,
                peeled: None,
                symref: Some(target.to_string()),
            }],
        )?;
        self.inner.set_head(target)
    }
    fn set_head_projected(&self, target: &str) -> anyhow::Result<TxId> {
        let _m = self.log.mutation();
        self.log_refs(
            false,
            vec![RefEdit {
                name: "HEAD".to_string(),
                old: None,
                new: None,
                peeled: None,
                symref: Some(target.to_string()),
            }],
        )?;
        self.inner.set_head_projected(target)
    }
    fn emit_pack(&self, objects: &[Oid<'_>], have: &[Oid<'_>], caps: &Caps, out: &mut dyn std::io::Write) -> anyhow::Result<PackStats> {
        self.inner.emit_pack(objects, have, caps, out)
    }
    fn select(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Option<ReachSet>> {
        self.inner.select(want, have)
    }
}

// ── control store ───────────────────────────────────────────────────────────

/// A `ControlStore` whose `put_blob` and `commit_doc` are frames first.
pub struct LoggedControlStore {
    inner: Arc<dyn ControlStore>,
    log: Arc<Log>,
}

impl LoggedControlStore {
    /// Wrap.
    pub fn new(inner: Arc<dyn ControlStore>, log: Arc<Log>) -> Self {
        Self { inner, log }
    }
    /// The wrapped store.
    pub fn inner(&self) -> &Arc<dyn ControlStore> {
        &self.inner
    }
}

impl ControlStore for LoggedControlStore {
    /// **A shard blob is framed and stored WITHOUT a flush of its own** (perf twin-apply-fsync,
    /// 2026-10-06). A content-addressed blob is part of no state until a document points at it,
    /// and the pointer move is ALWAYS a later frame: `commit_doc` appends its `ControlDoc` frame
    /// durably — that `fdatasync` covers every blob frame staged before it in the same file — and
    /// only then moves the pointer with a durable redb commit, which persists every blob row
    /// committed before it. So the log is still written before every state anybody can see, and
    /// a crash in between leaves at most an unreferenced blob nothing reaches. Measured
    /// (`tests/main_doc_write_io.rs`): one document change was five frames with ten flushes
    /// (five `fdatasync`s, five redb fsyncs); it is one of each.
    fn put_blob(&self, bytes: &[u8]) -> crate::Result<BlobId> {
        let id = BlobId::of(bytes);
        let meta = Meta::ControlBlob { id: blob_hex(&id) };
        let _m = self.log.mutation();
        self.log.stage(Kind::ControlBlob, &meta, bytes).map_err(log_err)?;
        self.inner.put_blob_unsynced(bytes)
    }
    fn get_blob(&self, id: &BlobId) -> crate::Result<Option<Vec<u8>>> {
        self.inner.get_blob(id)
    }
    fn read_doc(&self, name: &str) -> crate::Result<Option<BlobId>> {
        self.inner.read_doc(name)
    }
    fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> crate::Result<bool> {
        let meta = Meta::ControlDoc {
            name: name.to_string(),
            expected: expected.as_ref().map(blob_hex),
            new: blob_hex(&new),
        };
        let _m = self.log.mutation();
        self.log.append(Kind::ControlDoc, &meta, &[]).map_err(log_err)?;
        self.inner.commit_doc(name, expected, new)
    }
    fn history(&self, name: &str) -> crate::Result<Vec<DocVersion>> {
        self.inner.history(name)
    }
}

fn log_err(e: crate::txlog::Error) -> crate::Error {
    match e {
        // ★ grow-login-503: the log is RELEASED only while the data set is grown (the
        // runtime growth gave its files up for the unmount, `Log::release`, and takes them
        // back after the remount). A control write then is refused OFF THE STRIPE — the
        // control plane answers UNAVAILABLE ("retry") — never a control-store failure
        // (INTERNAL), which a console turned into "you are not authenticated" for a person
        // signing in during a disk grow (cov-E's round-3 KVM mockup, 2026-10-06).
        crate::txlog::Error::Released(_) => crate::Error::OffStripe {
            why: "the transaction log is released for the remount".into(),
        },
        // A box being restored (or a twin): its log follows; a write is "try again", never a
        // control-store failure (r5.1: "would not enrol gunnar-twin: … Internal" while main restored).
        crate::txlog::Error::Follower => crate::Error::LogFollows,
        e => crate::Error::ControlStore(format!("transaction log: {e}")),
    }
}

// ── secrets ─────────────────────────────────────────────────────────────────

/// A `SecretStore` whose `put` and `delete` are frames first. The blob is
/// the sealed bytes exactly as the store holds them.
pub struct LoggedSecretStore<S> {
    inner: S,
    log: Arc<Log>,
    /// Serialises `get_or_create` through this wrapper, so two creators of
    /// the same key never log two different values for one row.
    create: Mutex<()>,
}

impl<S: gunnar_store::SecretStore> LoggedSecretStore<S> {
    /// Wrap.
    pub fn new(inner: S, log: Arc<Log>) -> Self {
        Self {
            inner,
            log,
            create: Mutex::new(()),
        }
    }
    /// The wrapped store.
    pub fn inner(&self) -> &S {
        &self.inner
    }
    fn log_secret(&self, key: &str, value: &[u8], deleted: bool) -> gunnar_store::Result<()> {
        let meta = Meta::Secret {
            key: key.to_string(),
            deleted,
        };
        self.log
            .append(Kind::Secret, &meta, value)
            .map(|_| ())
            .map_err(|e| gunnar_store::Error::Io(std::io::Error::other(e.to_string())))
    }
}

impl<S: gunnar_store::SecretStore> gunnar_store::SecretStore for LoggedSecretStore<S> {
    fn get(&self, key: &str) -> gunnar_store::Result<Option<Vec<u8>>> {
        self.inner.get(key)
    }
    fn put(&self, key: &str, value: &[u8]) -> gunnar_store::Result<()> {
        let _m = self.log.mutation();
        self.log_secret(key, value, false)?;
        self.inner.put(key, value)
    }
    fn delete(&self, key: &str) -> gunnar_store::Result<()> {
        let _m = self.log.mutation();
        self.log_secret(key, &[], true)?;
        self.inner.delete(key)
    }
    fn get_or_create(&self, key: &str, mk: &mut dyn FnMut() -> gunnar_store::Result<Vec<u8>>) -> gunnar_store::Result<Vec<u8>> {
        let _one = self.create.lock().expect("secret create");
        if let Some(v) = self.inner.get(key)? {
            return Ok(v);
        }
        let value = mk()?;
        self.put(key, &value)?;
        Ok(value)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::txlog::Entry;
    use gunnar_store::SecretStore as _;

    fn log() -> (tempfile::TempDir, Arc<Log>) {
        let dir = tempfile::tempdir().unwrap();
        let log = Arc::new(Log::open(dir.path().join("txlog")).unwrap());
        (dir, log)
    }

    fn entries(log: &Log) -> Vec<Entry> {
        log.range(log.first_seq(), log.head_seq()).unwrap()
    }

    /// ★ grow-login-503 (cov-E's r3 KVM mockup, 2026-10-06): while main remounted after a
    /// growth, the kit's passkey login met `HTTP 401: internal error: the control store
    /// failed: transaction log: txlog: released …`. The log is RELEASED only while the data
    /// set is grown — a moment, not a fault — so a control write then is refused OFF THE
    /// STRIPE (UNAVAILABLE at every door: "retry"), never a control-store failure
    /// (INTERNAL, which the console turned into "you are not authenticated").
    #[test]
    fn a_control_write_while_the_log_is_released_is_off_the_stripe_never_a_store_failure() {
        let (_d, log) = log();
        let store = LoggedControlStore::new(Arc::new(crate::control::RedbControlStore::in_memory().unwrap()), Arc::clone(&log));
        log.release().unwrap();
        let refused = store.put_blob(b"{\"principals\":[]}").unwrap_err();
        assert!(matches!(refused, crate::Error::OffStripe { .. }), "{refused:?}");
        let said = refused.to_string();
        assert!(said.contains("the data set") && said.contains("remount"), "{said}");
        log.reacquire().unwrap();
        let id = store.put_blob(b"{\"principals\":[]}").expect("after the remount the write goes through");
        assert!(store.commit_doc("refs/gunnar/auth", None, id).unwrap());
    }

    #[test]
    fn a_control_write_is_a_blob_frame_then_a_doc_frame_before_the_store_sees_it() {
        let (_d, log) = log();
        let store = LoggedControlStore::new(Arc::new(crate::control::RedbControlStore::in_memory().unwrap()), Arc::clone(&log));
        let id = store.put_blob(b"{\"principals\":[]}").unwrap();
        assert!(store.commit_doc("refs/gunnar/auth", None, id).unwrap());
        // A lost CAS is logged with the claim, and the store answers false.
        let id2 = store.put_blob(b"{\"principals\":[1]}").unwrap();
        assert!(!store.commit_doc("refs/gunnar/auth", None, id2).unwrap());
        let e = entries(&log);
        assert_eq!(e.iter().map(|e| e.kind).collect::<Vec<_>>(), vec![3, 4, 3, 4]);
        match e[1].meta().unwrap() {
            Meta::ControlDoc { name, expected, new } => {
                assert_eq!(name, "refs/gunnar/auth");
                assert_eq!(expected, None);
                assert_eq!(new, blob_hex(&id));
            }
            other => panic!("{other:?}"),
        }
        assert_eq!(e[0].blob, b"{\"principals\":[]}");
    }

    /// **A shard blob rides on the document that publishes it** (perf, 2026-10-06): framed but
    /// not yet durable (nobody reads it, the head does not count it) until the `ControlDoc`
    /// frame's fsync makes both durable together — and a reopen from the disk finds both, in order.
    #[test]
    fn a_blob_frame_is_durable_with_the_document_frame_that_publishes_it() {
        let (d, log) = log();
        let store = LoggedControlStore::new(Arc::new(crate::control::RedbControlStore::in_memory().unwrap()), Arc::clone(&log));
        let syncs = log.frame_syncs();
        let id = store.put_blob(b"{\"principals\":[]}").unwrap();
        assert_eq!((log.head_seq(), log.unsynced()), (0, 1), "the blob frame waits for its document");
        assert!(store.commit_doc("refs/gunnar/auth", None, id).unwrap());
        assert_eq!((log.head_seq(), log.unsynced()), (2, 0), "durable together");
        assert_eq!(log.frame_syncs() - syncs, 1, "one fsync for the blob and its document");
        drop(store);
        drop(log);
        let reopened = Log::open(d.path().join("txlog")).unwrap();
        let kinds: Vec<u16> = (1..=reopened.head_seq()).map(|s| reopened.read(s).unwrap().kind).collect();
        assert_eq!(kinds, vec![Kind::ControlBlob as u16, Kind::ControlDoc as u16]);
    }

    #[test]
    fn a_secret_put_and_delete_are_frames_and_get_or_create_logs_once() {
        let (_d, log) = log();
        let store = LoggedSecretStore::new(gunnar_store::mem::MemSecretStore::new(), Arc::clone(&log));
        store.put("host_key", b"sealed").unwrap();
        let v = store.get_or_create("host_key", &mut || Ok(b"other".to_vec())).unwrap();
        assert_eq!(v, b"sealed");
        let c = store.get_or_create("issuer", &mut || Ok(b"minted".to_vec())).unwrap();
        assert_eq!(c, b"minted");
        store.delete("host_key").unwrap();
        let e = entries(&log);
        assert_eq!(e.len(), 3);
        assert_eq!(e[0].blob, b"sealed");
        assert_eq!(e[1].blob, b"minted");
        match e[2].meta().unwrap() {
            Meta::Secret { key, deleted } => {
                assert_eq!(key, "host_key");
                assert!(deleted);
            }
            other => panic!("{other:?}"),
        }
        assert_eq!(store.get("host_key").unwrap(), None);
    }
}