crates/gunnar-core/src/doc.rs
//! How the catalog and the identity table are actually stored.
//!
//! # Where this lives, and where it used to (the certificate-identity plan)
//!
//! A control table is a **JSON document in the [`ControlStore`], behind one
//! name**: content-addressed bytes, and a pointer moved by a compare-and-swap,
//! so two concurrent writers cannot lose each other's edits (the loser retries
//! against the new state).
//!
//! Until 2026-08-09 that store was a *git object database* and the pointer was
//! a git reference — the same shape, spelled in `ObjectStore` and `RefStore`.
//! It moved because the storage engine (the storage restart) takes objects **as packs**
//! and refuses a single-object write by design, which is exactly and only what
//! a control document is. `crate::control` has the full argument, including
//! what moving off git objects cost and what was done about each half.
//!
//! What did **not** move is everything below this paragraph: one document, one
//! name, one expectation, the same bounded retry, the same shard trie. The seam
//! is one trait wide, and swapping the thing behind it is the whole change.
//!
//! # What one document cost, and what replaced it
//!
//! Until 2026-08-03 the *whole* table was that one blob. Every read parsed all
//! of it and every write re-serialised all of it, so:
//!
//! * one SSH connection cost four full parses of the identity table plus one of
//! the catalog, before a byte of git protocol was exchanged (**measured**, on
//! a counted store, in `tests/control_cost.rs`); and
//! * creating one repository rewrote the catalog, which makes bulk creation
//! quadratic in the number of repositories.
//!
//! Two changes fixed that, and neither of them moved the seam above:
//!
//! 1. **[`DocRef`] caches the parsed document behind its version.** The
//! pointer is read on every call (one B-tree descent), and an unchanged blob
//! id means an unchanged document, so the parse happens once per *change*
//! rather than once per *read*. Another process's write moves the pointer
//! and is therefore seen; a cache that could not see it would not be a
//! cache, which is what `a_warm_handle_sees_another_writers_change` in
//! `tests/control_cost.rs` is there to catch.
//! 2. **[`ShardedMap`] splits every unbounded map out of the document** into a
//! fixed-arity trie of blobs: git's tree, one level shallower. The document
//! itself holds one blob id and a count per map, so it stays a few hundred
//! bytes at any scale, and a write touches one leaf, one branch and the root
//! node rather than the whole table.
//!
//! The compare-and-swap is untouched: one document, one name, one
//! expectation, the same bounded retry. Shard blobs are content-addressed, so a
//! losing writer's shards are simply unreferenced. There is no shard that can
//! be committed without the document that points at it.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex, MutexGuard};
use crate::control::{BlobId, ControlStore};
use crate::error::{Error, Result};
/// How many times a contended update is retried before giving up.
///
/// A retry only happens when another writer committed in between, and each one
/// re-reads and re-applies against the winner's state. The point of the bound
/// is that a livelock must surface as an error rather than as a hang.
pub const MAX_ATTEMPTS: usize = 8;
// ── WRITERS IN ONE PROCESS TAKE TURNS; THE COMPARE-AND-SWAP IS FOR STRANGERS ──
//
// Until 2026-08-29 there was no backoff and no writer lock, on this argument:
// redb serialises write transactions internally, so the only way to lose is on
// the expectation, and the winner has ALREADY COMMITTED, so the retry re-reads
// and makes progress immediately. True for two writers. False for eight.
//
// MEASURED 2026-08-29 (on the rig, then `tests/create_storm.rs` in-process):
// eight parallel push-creates lost `MAX_ATTEMPTS` compare-and-swaps in a row
// often enough to refuse 11 of 2 400 creates on the rig and **693 of 2 400** in
// one process with nothing but the catalog in the loop. Each loser re-reads,
// re-serialises the whole document, writes a blob and tries again while seven
// others do the same; with no pause the loser's next attempt lands in the
// same window as everyone else's, and whoever is unluckiest eight times in a
// row is told "the control plane is unavailable" for a race it could not lose
// by any fault of its own.
//
// Two things fix that, and both are here:
//
// 1. **[`DocRef::writers`]** — one mutex per document, shared by every clone of
// the handle, taken around the whole read-edit-commit. Every writer inside
// this process is now in a queue and a queue cannot livelock: each commit
// is one redb write transaction (~0.5 ms), so the eighth writer waits ~4 ms
// and never retries at all. Readers are untouched; `load` takes no lock.
// 2. **[`RETRY_BACKOFF`]** for the compare-and-swap that still can lose — a
// writer in ANOTHER process (there is none today: redb takes an exclusive
// file lock; but the CAS is the contract and the bound must be adequate on
// its own) — jittered so two losers do not retry in lockstep.
//
// The bound stays: a livelock must surface as an error rather than as a hang.
// It is simply no longer reachable from inside one process.
/// The pause before compare-and-swap attempt `n` (`n >= 2`): `n` ms, plus a
/// jitter of 0..`n` ms taken from the clock's low bits — no random number
/// generator is worth a dependency for this.
fn retry_backoff(attempt: usize) -> std::time::Duration {
let jitter = (crate::model::now_ms() as u64 % (attempt as u64 + 1)) as u64;
std::time::Duration::from_millis(attempt as u64 + jitter)
}
// ───────────────────────────── the sharded map ───────────────────────────────
/// Children per trie node. 32 × 32 = [`LEAVES`] leaves, reached in two hops.
///
/// The arithmetic this number comes from: a write costs one leaf plus two nodes
/// of `ARITY` entries each (about 75 bytes per entry with a sha256 id). At a
/// million repositories a leaf holds ~1000 records (call it 250 KB) and the
/// two nodes are 2.4 KB each, against the 300 MB the single-document layout
/// rewrote for the same create. Below about 30000 entries the nodes dominate
/// and the cost is flat at ~5 KB, which is the regime `tests/control_cost.rs`
/// measures.
const ARITY: usize = 32;
const LEAVES: usize = ARITY * ARITY;
/// Which leaf a key belongs in.
///
/// FNV-1a, written out rather than taken from `std`: `DefaultHasher` is
/// explicitly not stable across Rust releases, and a placement function that
/// changes under a compiler upgrade would leave every stored entry in a leaf
/// nobody looks in: a silent, total loss of the identity table.
fn leaf_of(key: &str) -> usize {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for byte in key.as_bytes() {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
(hash % LEAVES as u64) as usize
}
/// What a [`ShardedMap`] contributes to the enclosing document: the trie root
/// and the entry count, and nothing else.
///
/// `deny_unknown_fields` is load-bearing. The pre-shard layout put the entries
/// themselves here; without it, serde would ignore them and an identity table
/// full of principals would read as an empty one, which is the exact "a storage
/// fault must not look like an empty table" failure [`DocRef::read`] is written
/// against. With it, an unreadable layout is an error the operator sees.
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct StoredMap {
/// The trie's root node, absent when the map is empty.
#[serde(default, skip_serializing_if = "Option::is_none")]
root: Option<String>,
/// Total entries, maintained by the mutators so `counts()` is not a walk.
#[serde(default)]
len: usize,
}
/// The parts of a trie this handle has actually read.
struct Loaded<V> {
/// Level 0: branch slot → branch node id. `None` until first touched.
root: Option<Arc<BTreeMap<usize, BlobId>>>,
/// Level 1: branch slot → (leaf slot within the branch → leaf blob id).
branches: BTreeMap<usize, Arc<BTreeMap<usize, BlobId>>>,
/// Level 2: leaf slot → the entries in it.
leaves: BTreeMap<usize, Arc<BTreeMap<String, V>>>,
}
impl<V> Default for Loaded<V> {
fn default() -> Self {
Loaded {
root: None,
branches: BTreeMap::new(),
leaves: BTreeMap::new(),
}
}
}
impl<V> Clone for Loaded<V> {
/// Shallow: the loaded nodes are shared, the maps that index them are not.
///
/// That is what makes cloning a document cheap while still letting the
/// clone be mutated: a mutation copies the one leaf it touches and swaps in
/// a new `Arc`, leaving every other reader's view of that leaf alone. It is
/// copy-on-write, which is what [`DocRef::update`] needs: the cached
/// document must stay exactly as other threads see it until the commit
/// lands.
fn clone(&self) -> Self {
Loaded {
root: self.root.clone(),
branches: self.branches.clone(),
leaves: self.leaves.clone(),
}
}
}
/// A map whose entries live in their own blobs, addressed through a trie.
///
/// Reads load one leaf; writes rewrite one leaf, its branch and the root node.
/// Everything is content-addressed, so a leaf that did not change keeps its id
/// and the document that points at it is byte-identical, which is what lets
/// [`DocRef::update`] still recognise a no-op mutation.
///
/// Whole-map operations ([`ShardedMap::values`], [`ShardedMap::retain`]) read
/// every node and are O(entries) by construction. They are the listing and
/// tenant-deletion paths, never the connection path, and each one says so.
pub(crate) struct ShardedMap<V> {
stored: StoredMap,
store: Option<Arc<dyn ControlStore>>,
loaded: Mutex<Loaded<V>>,
/// Leaf slots mutated since the last [`ShardedMap::flush`].
dirty: BTreeSet<usize>,
}
impl<V> Default for ShardedMap<V> {
fn default() -> Self {
ShardedMap {
stored: StoredMap::default(),
store: None,
loaded: Mutex::new(Loaded::default()),
dirty: BTreeSet::new(),
}
}
}
impl<V> std::fmt::Debug for ShardedMap<V> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShardedMap")
.field("len", &self.stored.len)
.field("root", &self.stored.root)
.finish()
}
}
impl<V> Clone for ShardedMap<V> {
fn clone(&self) -> Self {
ShardedMap {
stored: self.stored.clone(),
store: self.store.clone(),
loaded: Mutex::new(guard(&self.loaded).clone()),
dirty: self.dirty.clone(),
}
}
}
impl<V> serde::Serialize for ShardedMap<V> {
fn serialize<S: serde::Serializer>(
&self,
serializer: S,
) -> std::result::Result<S::Ok, S::Error> {
// A document serialised with shards still in memory would point at the
// *previous* leaf and silently drop the edit. Nothing may write one.
if !self.dirty.is_empty() {
return Err(serde::ser::Error::custom(
"a sharded map was serialised with unwritten shards; Doc::flush was not called",
));
}
self.stored.serialize(serializer)
}
}
impl<'de, V> serde::Deserialize<'de> for ShardedMap<V> {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
Ok(ShardedMap {
stored: StoredMap::deserialize(deserializer)?,
..ShardedMap::default()
})
}
}
/// Take a lock whose only content is a cache, recovering from poisoning.
///
/// A panic under one of these leaves a *cache* behind, not a half-applied
/// mutation: every entry stored under it is a complete, already-committed
/// value. Propagating the poison would turn one panic anywhere in the process
/// into a permanently unauthenticatable server, which is a far worse failure
/// than the one being guarded against (the handle cache makes the same call).
fn guard<T>(lock: &Mutex<T>) -> MutexGuard<'_, T> {
lock.lock().unwrap_or_else(|poison| poison.into_inner())
}
fn node_bytes(node: &BTreeMap<usize, BlobId>) -> Result<Vec<u8>> {
let encoded: BTreeMap<String, String> = node
.iter()
.map(|(slot, id)| (slot.to_string(), id.to_string()))
.collect();
Ok(serde_json::to_vec(&encoded)?)
}
fn parse_node(bytes: &[u8], id: &BlobId) -> Result<BTreeMap<usize, BlobId>> {
let encoded: BTreeMap<String, String> =
serde_json::from_slice(bytes).map_err(|err| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("node {id} is not a trie node: {err}"),
})?;
let mut out = BTreeMap::new();
for (slot, child) in encoded {
let slot: usize = slot.parse().map_err(|_| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("node {id} has a non-numeric slot {slot:?}"),
})?;
let child = BlobId::from_hex(&child).ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("node {id} slot {slot} is not a blob id: {child:?}"),
})?;
out.insert(slot, child);
}
Ok(out)
}
fn read_blob<T: serde::de::DeserializeOwned>(
store: &dyn ControlStore,
id: &BlobId,
what: &str,
) -> Result<T> {
let bytes = store.get_blob(id)?.ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("{what} {id} is not in the control store"),
})?;
serde_json::from_slice(&bytes).map_err(|err| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("{what} {id} could not be read: {err}"),
})
}
impl<V> ShardedMap<V>
where
V: Clone + serde::Serialize + serde::de::DeserializeOwned,
{
fn store(&self) -> Result<&Arc<dyn ControlStore>> {
self.store.as_ref().ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: "a sharded map was used before it was bound to a control store".to_owned(),
})
}
/// Hand this map the store its shards live in. Called by [`Doc::bind`].
pub(crate) fn bind(&mut self, store: &Arc<dyn ControlStore>) {
self.store = Some(Arc::clone(store));
}
fn root_node(
loaded: &mut Loaded<V>,
stored: &StoredMap,
store: &dyn ControlStore,
) -> Result<Arc<BTreeMap<usize, BlobId>>> {
if let Some(root) = &loaded.root {
return Ok(Arc::clone(root));
}
let node = match &stored.root {
None => BTreeMap::new(),
Some(hex) => {
let id = BlobId::from_hex(hex).ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("the trie root {hex:?} is not a blob id"),
})?;
let bytes = store.get_blob(&id)?.ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("the trie root {id} is not in the control store"),
})?;
parse_node(&bytes, &id)?
}
};
let node = Arc::new(node);
loaded.root = Some(Arc::clone(&node));
Ok(node)
}
fn branch_node(
loaded: &mut Loaded<V>,
stored: &StoredMap,
store: &dyn ControlStore,
branch: usize,
) -> Result<Arc<BTreeMap<usize, BlobId>>> {
if let Some(node) = loaded.branches.get(&branch) {
return Ok(Arc::clone(node));
}
let root = Self::root_node(loaded, stored, store)?;
let node = match root.get(&branch) {
None => BTreeMap::new(),
Some(id) => {
let bytes = store.get_blob(id)?.ok_or_else(|| Error::CorruptDocument {
ref_name: "<shard trie>".to_owned(),
reason: format!("trie branch {id} is not in the control store"),
})?;
parse_node(&bytes, id)?
}
};
let node = Arc::new(node);
loaded.branches.insert(branch, Arc::clone(&node));
Ok(node)
}
fn leaf_map(
loaded: &mut Loaded<V>,
stored: &StoredMap,
store: &dyn ControlStore,
slot: usize,
) -> Result<Arc<BTreeMap<String, V>>> {
if let Some(leaf) = loaded.leaves.get(&slot) {
return Ok(Arc::clone(leaf));
}
let branch = Self::branch_node(loaded, stored, store, slot / ARITY)?;
let leaf = match branch.get(&(slot % ARITY)) {
None => BTreeMap::new(),
Some(id) => read_blob(store, id, "trie leaf")?,
};
let leaf = Arc::new(leaf);
loaded.leaves.insert(slot, Arc::clone(&leaf));
Ok(leaf)
}
fn leaf(&self, slot: usize) -> Result<Arc<BTreeMap<String, V>>> {
let store = Arc::clone(self.store()?);
let mut loaded = guard(&self.loaded);
Self::leaf_map(&mut loaded, &self.stored, store.as_ref(), slot)
}
/// How many entries there are. Maintained on write, so this is not a walk.
pub(crate) fn len(&self) -> usize {
self.stored.len
}
pub(crate) fn is_empty(&self) -> bool {
self.stored.len == 0
}
/// One entry. Reads at most three blobs, and none once they are cached.
pub(crate) fn get(&self, key: &str) -> Result<Option<V>> {
Ok(self.leaf(leaf_of(key))?.get(key).cloned())
}
pub(crate) fn contains_key(&self, key: &str) -> Result<bool> {
Ok(self.leaf(leaf_of(key))?.contains_key(key))
}
/// Insert, returning what was there. Same cost as [`ShardedMap::get`] plus
/// the write, which [`Doc::flush`] does once for the whole mutation.
pub(crate) fn insert(&mut self, key: String, value: V) -> Result<Option<V>> {
let slot = leaf_of(&key);
let leaf = self.leaf(slot)?;
let mut next = (*leaf).clone();
let previous = next.insert(key, value);
if previous.is_none() {
self.stored.len += 1;
}
guard(&self.loaded).leaves.insert(slot, Arc::new(next));
self.dirty.insert(slot);
Ok(previous)
}
pub(crate) fn remove(&mut self, key: &str) -> Result<Option<V>> {
let slot = leaf_of(key);
let leaf = self.leaf(slot)?;
if !leaf.contains_key(key) {
return Ok(None);
}
let mut next = (*leaf).clone();
let previous = next.remove(key);
self.stored.len -= 1;
guard(&self.loaded).leaves.insert(slot, Arc::new(next));
self.dirty.insert(slot);
Ok(previous)
}
/// Change one entry in place. `false` if it is not there.
pub(crate) fn modify(&mut self, key: &str, edit: impl FnOnce(&mut V)) -> Result<bool> {
let slot = leaf_of(key);
let leaf = self.leaf(slot)?;
if !leaf.contains_key(key) {
return Ok(false);
}
let mut next = (*leaf).clone();
if let Some(value) = next.get_mut(key) {
edit(value);
}
guard(&self.loaded).leaves.insert(slot, Arc::new(next));
self.dirty.insert(slot);
Ok(true)
}
/// Every entry, key and value, in key order.
///
/// **O(entries)**. Reads every node in the trie. This is the listing path
/// (`repo list`, `principal list`), never the connection path.
pub(crate) fn entries(&self) -> Result<BTreeMap<String, V>> {
// The leaves are gathered under the lock and WALKED outside it
// (round 5.3.1, test5 2026-10-08: `Repos.List` p50 2.7 s at 96 rps on a
// half-idle main — every listing cloned the whole catalog while
// holding the one cache lock every other reader of it waits for).
let mut out = BTreeMap::new();
for leaf in self.all_leaves()? {
for (key, value) in leaf.iter() {
out.insert(key.clone(), value.clone());
}
}
Ok(out)
}
/// **Every leaf, loaded** — the shared, immutable `Arc`s, gathered under
/// the cache lock (which also loads what was never read) and handed back
/// so the caller walks them with the lock released.
fn all_leaves(&self) -> Result<Vec<Arc<BTreeMap<String, V>>>> {
let store = Arc::clone(self.store()?);
let mut loaded = guard(&self.loaded);
let root = Self::root_node(&mut loaded, &self.stored, store.as_ref())?;
let branches: Vec<usize> = root.keys().copied().collect();
let mut leaves = Vec::new();
for branch in branches {
let node = Self::branch_node(&mut loaded, &self.stored, store.as_ref(), branch)?;
for child in node.keys().copied().collect::<Vec<_>>() {
leaves.push(Self::leaf_map(
&mut loaded,
&self.stored,
store.as_ref(),
branch * ARITY + child,
)?);
}
}
Ok(leaves)
}
/// **Is there an entry `hit` accepts?** The walk of [`ShardedMap::entries`], stopping at the
/// first hit and cloning nothing — for a yes/no a caller asks on every call
/// (`AuthTable::is_enterprise`, `Access.WhoAmI`). **O(entries)** at worst.
pub(crate) fn any(&self, mut hit: impl FnMut(&str, &V) -> bool) -> Result<bool> {
// The test runs with the cache lock released (see `entries`).
for leaf in self.all_leaves()? {
if leaf.iter().any(|(key, value)| hit(key, value)) {
return Ok(true);
}
}
Ok(false)
}
/// Every value, in key order. **O(entries)**, as [`ShardedMap::entries`].
pub(crate) fn values(&self) -> Result<Vec<V>> {
Ok(self.entries()?.into_values().collect())
}
/// Every key, in order. **O(entries)**, as [`ShardedMap::entries`] — and it
/// pays the same price, because the trie stores keys in the leaves beside
/// their values and there is no cheaper walk. It exists so that a caller
/// who wants only the names (`FlowStore::namespaces`) does not have to
/// clone every value to get them.
pub(crate) fn keys(&self) -> Result<Vec<String>> {
Ok(self.entries()?.into_keys().collect())
}
/// Drop every entry `keep` rejects. **O(entries)**, and the reason
/// `remove_principal` costs the whole key table; see its own note.
pub(crate) fn retain(&mut self, mut keep: impl FnMut(&str, &V) -> bool) -> Result<()> {
let doomed: Vec<String> = self
.entries()?
.into_iter()
.filter(|(key, value)| !keep(key, value))
.map(|(key, _)| key)
.collect();
for key in doomed {
self.remove(&key)?;
}
Ok(())
}
/// Write every leaf a mutation touched, then the branches above them, then
/// the root, and leave the enclosing document pointing at the new root.
pub(crate) fn flush(&mut self) -> Result<()> {
if self.dirty.is_empty() {
return Ok(());
}
let store = Arc::clone(self.store()?);
let store = store.as_ref();
let mut loaded = guard(&self.loaded);
let mut root = (*Self::root_node(&mut loaded, &self.stored, store)?).clone();
let branches: BTreeSet<usize> = self.dirty.iter().map(|slot| slot / ARITY).collect();
for branch in branches {
let mut node = (*Self::branch_node(&mut loaded, &self.stored, store, branch)?).clone();
for slot in self
.dirty
.iter()
.copied()
.filter(|slot| slot / ARITY == branch)
{
let leaf = Self::leaf_map(&mut loaded, &self.stored, store, slot)?;
if leaf.is_empty() {
node.remove(&(slot % ARITY));
} else {
let bytes = serde_json::to_vec(&*leaf)?;
let id = store.put_blob(&bytes)?;
node.insert(slot % ARITY, id);
}
}
if node.is_empty() {
root.remove(&branch);
} else {
let id = store.put_blob(&node_bytes(&node)?)?;
root.insert(branch, id);
}
loaded.branches.insert(branch, Arc::new(node));
}
self.stored.root = if root.is_empty() {
None
} else {
let id = store.put_blob(&node_bytes(&root)?)?;
Some(id.to_string())
};
loaded.root = Some(Arc::new(root));
drop(loaded);
self.dirty.clear();
Ok(())
}
}
// ───────────────────────────── the document ──────────────────────────────────
/// A control document: a serde value with sharded maps inside it.
///
/// The two hooks exist because the maps cannot reach the control store through
/// serde. `bind` gives every map inside the document the store it was read
/// from; `flush` writes the shards a mutation touched, so that by the time the
/// document is serialised its trie roots are real blob ids.
pub(crate) trait Doc:
serde::Serialize + serde::de::DeserializeOwned + Default + Clone + Send + Sync + 'static
{
/// Hand every sharded map inside this document its control store.
fn bind(&mut self, store: &Arc<dyn ControlStore>);
/// Write every shard this document's mutation touched.
fn flush(&mut self) -> Result<()>;
}
/// Bind and flush a document's sharded maps.
///
/// Both halves of [`Doc`] are mechanical and identical for every field, so
/// `impl_doc!` writes them rather than each document repeating them, the same
/// reason [`crate::sandbox`] has one resolver (one implementation). Adding a field to a
/// document and forgetting to bind it would produce a map that errors on first
/// use; listing the fields once is what stops that being two places.
macro_rules! impl_doc {
($doc:ty { $($field:ident),+ $(,)? }) => {
impl crate::doc::Doc for $doc {
fn bind(&mut self, store: &std::sync::Arc<dyn crate::control::ControlStore>) {
$(self.$field.bind(store);)+
}
fn flush(&mut self) -> crate::error::Result<()> {
$(self.$field.flush()?;)+
Ok(())
}
}
};
}
pub(crate) use impl_doc;
/// The parsed document, and the version it was parsed from.
struct Cached<T> {
version: Option<BlobId>,
doc: Arc<T>,
}
/// A serde document stored as a content-addressed blob, pointed at by one name
/// in the [`ControlStore`].
pub(crate) struct DocRef<T> {
store: Arc<dyn ControlStore>,
/// The document's name in the store. Still `refs/gunnar/auth` and
/// `refs/gunnar/catalog`: they no longer name git references, but they are
/// what every error message, log line and operator runbook calls these two
/// documents, and renaming them would have been a second change riding on
/// this one.
///
/// A `Cow` since 2026-10-05: an account's audit log is one document PER
/// ACCOUNT (`refs/gunnar/account-audit/<account>`), named at run time.
name: std::borrow::Cow<'static, str>,
/// Shared by every clone of this handle, because every clone is the same
/// document. Two handles in one process caching separately would parse
/// twice for no reason.
cache: Arc<Mutex<Option<Cached<T>>>>,
/// **The writers' queue.** Held across one whole [`DocRef::update`], and
/// shared by every clone for the same reason the cache is: every clone is
/// the same document, and two writers on it must take turns rather than
/// race the compare-and-swap. See the note beside [`MAX_ATTEMPTS`].
writers: Arc<Mutex<()>>,
}
impl<T> std::fmt::Debug for DocRef<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DocRef").field("name", &self.name).finish()
}
}
impl<T> Clone for DocRef<T> {
fn clone(&self) -> Self {
DocRef {
store: Arc::clone(&self.store),
name: self.name.clone(),
cache: Arc::clone(&self.cache),
writers: Arc::clone(&self.writers),
}
}
}
impl<T: Doc> DocRef<T> {
pub(crate) fn new(store: Arc<dyn ControlStore>, name: &'static str) -> Result<Self> {
Self::named(store, std::borrow::Cow::Borrowed(name))
}
/// A document whose name is made at run time (one per account).
pub(crate) fn named(
store: Arc<dyn ControlStore>,
name: std::borrow::Cow<'static, str>,
) -> Result<Self> {
Ok(DocRef {
store,
name,
cache: Arc::new(Mutex::new(None)),
writers: Arc::new(Mutex::new(())),
})
}
/// The name this document lives under.
pub(crate) fn ref_name(&self) -> &str {
&self.name
}
/// **The document's current version** — the blob its pointer names, `None`
/// for a document never written. One pointer read and no parse: what a
/// reader that keeps its own index of the document asks to learn whether
/// it must rebuild (gunnar-server's usage ledger, 2026-10-06).
pub(crate) fn version(&self) -> Result<Option<crate::control::BlobId>> {
self.store.read_doc(&self.name)
}
/// **Every version this document has ever had**, oldest first (the certificate-identity plan).
///
/// This is what a ref update per change used to give away. It is not free
/// on a key/value store, so it is written on purpose — in the same
/// transaction as the pointer move, so the trail cannot disagree with the
/// document it describes.
pub(crate) fn history(&self) -> Result<Vec<crate::control::DocVersion>> {
self.store.history(&self.name)
}
fn empty(&self) -> T {
let mut doc = T::default();
doc.bind(&self.store);
doc
}
/// Read the document, and the version it was read from.
///
/// A document that has never been written means an empty table, which is a
/// legitimate state (a fresh server). A pointer at something unreadable
/// does **not** mean an empty table and is an error. The two must never
/// look the same, or a storage fault reads as "no principals are
/// registered", which is exactly the failure that opens a server up.
///
/// The pointer is read every time. That is the whole basis of the cache:
/// the blob id IS the document's version, so an unchanged id is an
/// unchanged document, including when the writer was another process.
fn read(&self) -> Result<(Option<BlobId>, Arc<T>)> {
let version = self.store.read_doc(&self.name)?;
if let Some(cached) = guard(&self.cache).as_ref() {
if cached.version == version {
return Ok((version, Arc::clone(&cached.doc)));
}
}
let doc = match &version {
None => self.empty(),
Some(id) => {
let bytes = self.store.get_blob(id)?.ok_or_else(|| {
// The pointer names a version the store does not hold. On a
// git object database this was a dangling reference; here it
// is a blob table that lost a row. Either way it is a FAULT
// and must not read as an empty table.
Error::CorruptDocument {
ref_name: self.name.to_string(),
reason: format!("points at {id}, which is not in the control store"),
}
})?;
let mut doc: T =
serde_json::from_slice(&bytes).map_err(|err| Error::CorruptDocument {
ref_name: self.name.to_string(),
reason: format!("blob {id} is not a valid document: {err}"),
})?;
doc.bind(&self.store);
doc
}
};
let doc = Arc::new(doc);
self.remember(version, Arc::clone(&doc));
Ok((version, doc))
}
fn remember(&self, version: Option<BlobId>, doc: Arc<T>) {
*guard(&self.cache) = Some(Cached { version, doc });
}
/// The current document.
///
/// Shared, not copied: callers read through it and clone the one row they
/// want. In steady state this costs one pointer read and no parse.
pub(crate) fn load(&self) -> Result<Arc<T>> {
Ok(self.read()?.1)
}
/// Read, modify, and commit with a compare-and-swap, retrying if another
/// writer got there first.
///
/// `edit` may run more than once, so it must decide from the state it is
/// handed, not from anything captured earlier. It returns the caller's
/// value, so a mutation can hand back the row it created without a second
/// read. Each attempt starts from a fresh [`DocRef::read`], so a retry sees
/// the winner's state and not a stale copy of its own.
pub(crate) fn update<R>(&self, mut edit: impl FnMut(&mut T) -> Result<R>) -> Result<R> {
// The document commit, timed and counted. Contention on a
// control document is one of the two things the fsync ceiling is about
// and it
// was completely invisible: a retry logged one `debug!` with no
// document, no attempt number and no duration, so "the control plane
// is slow" had nothing behind it.
let span = tracing::debug_span!(
target: crate::observe::TARGET_STORE,
"store.doc_txn",
document = &*self.name,
);
let _entered = span.enter();
let started = std::time::Instant::now();
// Writers take turns. A poisoned lock means a writer panicked inside
// its `edit`; nothing under the lock is state (it guards a turn, not
// data), so the next writer simply takes its turn.
let _turn = self
.writers
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let queued_ms = started.elapsed().as_millis() as u64;
for attempt in 1..=MAX_ATTEMPTS {
let (previous, current) = self.read()?;
// Copy-on-write: the cached document stays exactly as every other
// reader sees it until this writer's commit lands.
let mut doc = (*current).clone();
let out = edit(&mut doc)?;
// Shards first. The document about to be written must point at
// blobs that are already in the store, or a reader that follows the
// pointer would find a dangling trie.
doc.flush()?;
// Deterministic encoding: every map in the document is ordered and
// every shard is content-addressed, so the same table always hashes
// to the same blob and a no-op update is visible as "the pointer did
// not move" rather than as a new version.
let bytes = serde_json::to_vec(&doc)?;
let new = self.store.put_blob(&bytes)?;
if previous == Some(new) {
// A no-op update is a *fact*, not a silence: it is what an
// idempotent `grant` or a re-registered key looks like, and an
// operator chasing "my change did not take" needs to be able
// to tell it from a write that failed. It also keeps the audit
// trail honest — a version that changed nothing is not a
// version, and appending one would make `history` a log of
// reads.
tracing::debug!(
target: crate::observe::TARGET_STORE,
attempt,
elapsed_ms = started.elapsed().as_millis() as u64,
outcome = "unchanged",
"control document rewrote to the same bytes; the pointer was not moved"
);
return Ok(out);
}
// ONE failure mode, where the reference store had two. redb waits
// inside `begin_write` rather than refusing a second committer, so
// "somebody holds the lock" is not an answer a caller can get and
// `Ok(false)` means exactly one thing: the winner already committed.
// (MEASURED 2026-08-05, on the shape this replaced: two concurrent
// `Bootstrap.Enroll` calls produced one winner and one "a lock could
// not be obtained for reference refs/gunnar/auth" — the right
// OUTCOME reported as a storage fault.)
if self.store.commit_doc(&self.name, previous, new)? {
self.remember(Some(new), Arc::new(doc));
tracing::debug!(
target: crate::observe::TARGET_STORE,
attempt,
queued_ms,
elapsed_ms = started.elapsed().as_millis() as u64,
bytes = bytes.len(),
outcome = "committed",
"control document committed"
);
return Ok(out);
}
// `warn!`, not `debug!`, from the second retry on: one lost race is
// normal, a document that keeps losing them is the writer contention
// the fsync ceiling is about and it must not take a debug build to see.
if attempt >= 2 {
tracing::warn!(
target: crate::observe::TARGET_STORE,
attempt,
elapsed_ms = started.elapsed().as_millis() as u64,
"control document changed under us AGAIN; retrying"
);
} else {
tracing::debug!(
target: crate::observe::TARGET_STORE,
attempt,
"control document changed under us; retrying"
);
}
// Somebody outside this process's writer queue committed under us.
// Back off with jitter (see the note beside `MAX_ATTEMPTS`) so the
// next attempt does not land in the same window again.
std::thread::sleep(retry_backoff(attempt + 1));
}
tracing::error!(
target: crate::observe::TARGET_STORE,
attempts = MAX_ATTEMPTS,
elapsed_ms = started.elapsed().as_millis() as u64,
"the control document lost every compare-and-swap; giving up"
);
Err(Error::Contended {
ref_name: self.name.to_string(),
attempts: MAX_ATTEMPTS,
})
}
}