crates/gunnar-core/src/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);
}
}