crates/gunnar-core/src/control.rs
//! **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 ©.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"[..])
);
}
}