gunnar-core 3.0.0

//! 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,
        })
    }
}