Skip to main content

gunnar_core/
control.rs

1//! **Where the control plane's own state lives** — the catalog, the identity
2//! table, and every document [`crate::doc`] puts behind a name.
3//!
4//! # Why this is not the object store any more (the certificate-identity plan)
5//!
6//! Until 2026-08-09 a control document was a blob in the repository object
7//! store with a reference pointing at it — gunnar's own `ObjectStore` and
8//! `RefStore` traits, both since deleted in favour of `GitOps`: git's own
9//! trick, and defensible while the only durable backend gunnar had *was* a git
10//! directory. It stopped being defensible when the storage engine landed
11//! (the storage restart). That engine takes objects **as packs** — `GitOps::put_pack` is
12//! where the closure check and the durability ordering live — and it refuses a
13//! single-object write by design. So `principal add` and `namespace create`
14//! both died before they reached any of their own logic, with
15//!
16//! ```text
17//!   the znippy-git store takes objects as packs, not one at a time
18//! ```
19//!
20//! The fix is not a pack wrapper around a 200-byte JSON blob. User records were
21//! never git objects for a *reason*; they were git objects because the control
22//! plane predates anywhere better. This is that better place: an embedded
23//! key/value store with real transactions, which is what a compare-and-swap on
24//! a document wanted all along.
25//!
26//! redb adds nothing to the appliance that was not already there — it is the
27//! znippy engine's own tail — and it has no build script beyond a version probe
28//! and no dependencies of its own.
29//!
30//! # What is stored, and in which table
31//!
32//! ```text
33//!   documents          name            → 32-byte blob id      the current version
34//!   blobs              32-byte id      → bytes                content-addressed
35//!   document_history   (name, seq)     → millis ‖ blob id     append-only
36//!   secrets            key             → tag ‖ sealed bytes   crate::secrets
37//! ```
38//!
39//! The fourth table arrived on 2026-08-10 and finished the migration this one
40//! started: the SSH host key and the issuer private key were still in an Arrow
41//! log inside `control.git`, which is what kept `gunnar-server` opening a
42//! repository — and therefore needing a storage engine — in order to boot. It
43//! is a *table* here rather than a document because a secret is not a document:
44//! there is no compare-and-swap on it, no history worth keeping (a superseded
45//! host key is the defect, not the audit trail) and no content addressing to
46//! want. See [`crate::secrets`], which owns everything about its shape.
47//!
48//! **Content addressing is kept**, and not out of nostalgia for git. Three
49//! properties of [`crate::doc`] rest on it and would each have to be rebuilt
50//! without it: a shard that did not change keeps its id, so a document that did
51//! not change hashes to the same bytes and a no-op update is visible as *the
52//! pointer did not move* rather than as a new version; a losing writer's shards
53//! are simply unreferenced rather than needing a rollback; and a reader that
54//! follows a document to its shards can never find a leaf that was rewritten
55//! underneath it.
56//!
57//! # The two things moving off git objects cost (the certificate-identity plan)
58//!
59//! * **Replication.** Metadata used to replicate *as a repository*, by the same
60//!   mechanism as everything else: `control.git` mirrored, and the catalog and
61//!   identity table came with it. It does not any more. `<data-root>/control.redb`
62//!   is a file a standby has to be given some other way, and there is no
63//!   mechanism here that gives it one. **Named, not rebuilt.**
64//! * **History.** A ref update per change was a free audit trail: every previous
65//!   version of the identity table stayed in the object store with the reflog
66//!   ordering them. That is *not* free on a key/value store, so it is explicit:
67//!   [`DOC_HISTORY`] is appended in the same write transaction as the pointer
68//!   move, and [`ControlStore::history`] reads it back. Rebuilt, deliberately —
69//!   an ACL whose changes cannot be listed is an ACL nobody can audit, and the
70//!   whole table is one B-tree append per change.
71//!
72//! Nothing prunes either table. Old document versions and superseded shards
73//! accumulate exactly as loose objects did, which is the same bargain git makes
74//! and the same one a `gunnar control gc` would have to unmake.
75
76use std::path::Path;
77use std::sync::Arc;
78
79use redb::ReadableTable;
80
81use crate::error::{Error, Result};
82
83/// The current version of every named document.
84const DOCUMENTS: redb::TableDefinition<'static, &str, &[u8]> =
85    redb::TableDefinition::new("documents");
86
87/// Content-addressed bytes: documents themselves, and the shard blobs they
88/// point at.
89const BLOBS: redb::TableDefinition<'static, &[u8], &[u8]> = redb::TableDefinition::new("blobs");
90
91/// **The audit trail** (the certificate-identity plan): every version a
92/// document has ever had, in the order it had them, written in the same
93/// transaction as the pointer move.
94///
95/// Keyed `(document name, sequence)` so one document's history is one range
96/// scan and two documents cannot interleave. The value is the commit time in
97/// milliseconds since the epoch, big-endian, followed by the blob id.
98const DOC_HISTORY: redb::TableDefinition<'static, (&str, u64), &[u8]> =
99    redb::TableDefinition::new("document_history");
100
101/// **What this box has applied, by sequence number** — per box, never framed, never in a
102/// snapshot or an export (perf twin-apply-fsync, 2026-10-06). One row today: [`TWIN_APPLIED`].
103const MARKS: redb::TableDefinition<'static, &str, u64> = redb::TableDefinition::new("box_marks");
104
105/// **When each principal last signed in and last changed something** (owner,
106/// 2026-10-21) — principal id → its stamps ([`crate::activity`], JSON). Per box
107/// like [`MARKS`]: never framed, never in a snapshot or an export, and never in
108/// the identity document. Written by the activity flusher only, one transaction
109/// per flush that moved something.
110const PRINCIPAL_ACTIVITY: redb::TableDefinition<'static, &str, &[u8]> =
111    redb::TableDefinition::new("principal_activity");
112
113/// **The last transaction-log entry whose control-plane writes are in this database** — a
114/// twin's applied watermark. Committed in the SAME redb transaction as the writes it vouches
115/// for ([`ControlDb::commit_group`]), so after a crash the database is exactly the state through
116/// the marker, and the entries after it are re-applied in order from there. A twin used to re-apply
117/// only its log's head, which is exact only when every entry is made durable on its own; a
118/// group commit needs the marker.
119pub const TWIN_APPLIED: &str = "twin.applied";
120
121/// **What a write closure decided** ([`ControlDb::write`]): commit what it wrote, or discard
122/// the transaction — which a closure may only do when it wrote NOTHING (a lost compare-and-swap,
123/// a secret that already exists), because inside a group ([`ControlDb::begin_group`]) a discard
124/// cannot take back what the transaction already holds.
125pub(crate) enum Txn<T> {
126    /// Keep the writes.
127    Commit(T),
128    /// Keep the writes WITHOUT an fsync of their own: durable with the next durable commit
129    /// ([`ControlStore::put_blob_unsynced`]).
130    CommitUnsynced(T),
131    /// Nothing was written; nothing to keep.
132    Discard(T),
133}
134
135/// **One redb write transaction shared by a run of writes on ONE thread** (perf
136/// twin-apply-fsync, 2026-10-06): a twin takes a whole group of log entries into it and commits
137/// it once — each B-tree page it touches is written once, and the group costs one fsync.
138struct Group {
139    owner: std::thread::ThreadId,
140    txn: redb::WriteTransaction,
141    /// A write committed into it.
142    dirty: bool,
143    /// A write failed half-way: the group is aborted whole at its end.
144    poisoned: bool,
145}
146
147/// How many bytes a [`BlobId`] is.
148pub const BLOB_ID_LEN: usize = 32;
149
150/// A content address: sha-256 over the stored bytes.
151///
152/// Not a `gix_hash::ObjectId`. The old id was a *git* object id because the
153/// bytes were in a git object database and git had to be able to name them.
154/// Nothing here is git-visible, so the id is the hash and nothing else — no
155/// object kind, no `blob <len>\0` header, and no hash-kind negotiation with a
156/// repository that is not involved.
157#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
158pub struct BlobId([u8; BLOB_ID_LEN]);
159
160impl BlobId {
161    /// The address of these bytes.
162    pub fn of(bytes: &[u8]) -> Self {
163        use sha2::Digest;
164        let mut hasher = sha2::Sha256::new();
165        hasher.update(bytes);
166        BlobId(hasher.finalize().into())
167    }
168
169    /// The raw address.
170    pub fn as_bytes(&self) -> &[u8; BLOB_ID_LEN] {
171        &self.0
172    }
173
174    /// Parse the hex form. `None` if it is not [`BLOB_ID_LEN`] bytes of hex.
175    pub fn from_hex(text: &str) -> Option<Self> {
176        let mut out = [0u8; BLOB_ID_LEN];
177        hex::decode_to_slice(text, &mut out).ok()?;
178        Some(BlobId(out))
179    }
180
181    /// Rebuild from raw bytes, as they come back out of a table.
182    fn from_slice(bytes: &[u8]) -> Option<Self> {
183        Some(BlobId(bytes.try_into().ok()?))
184    }
185}
186
187impl std::fmt::Display for BlobId {
188    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
189        f.write_str(&hex::encode(self.0))
190    }
191}
192
193impl std::fmt::Debug for BlobId {
194    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195        write!(f, "BlobId({self})")
196    }
197}
198
199/// One version a document has had, out of [`ControlStore::history`].
200#[derive(Clone, PartialEq, Eq, Debug)]
201pub struct DocVersion {
202    /// Its position in the document's history, from 1.
203    pub seq: u64,
204    /// When it was committed, in milliseconds since the unix epoch.
205    pub at: crate::model::UnixMillis,
206    /// The version itself, still readable through
207    /// [`ControlStore::get_blob`] — nothing prunes the blob table.
208    pub blob: BlobId,
209}
210
211/// **Where a control document is kept.**
212///
213/// A trait rather than the concrete store, for one reason, and it is measure,
214/// never assert:
215/// `tests/control_cost.rs` proves the compare-and-swap by running the same
216/// concurrency test against a store that *ignores the expectation*, in the same
217/// run, and asserting that acknowledged writes are lost. A guard that cannot be
218/// broken is a guard nobody has seen fail. The counting decorator that measures
219/// what a connection costs lives on the same seam.
220pub trait ControlStore: Send + Sync + 'static {
221    /// Store these bytes and return their address. Idempotent: storing the same
222    /// bytes twice is one row and the same id.
223    fn put_blob(&self, bytes: &[u8]) -> Result<BlobId>;
224
225    /// **[`ControlStore::put_blob`] for a blob that only a LATER [`ControlStore::commit_doc`]
226    /// makes part of any state** (perf twin-apply-fsync, 2026-10-06): the store may keep it
227    /// without making it durable on its own, because the pointer move that publishes it is
228    /// durable and persists it — redb's durable commit persists every earlier commit. Until then
229    /// it is an unreferenced, content-addressed row nothing can reach, and a crash that takes it
230    /// takes nothing anybody could see. The default is the durable `put_blob`.
231    ///
232    /// The primary's tee uses it for a document's shards. A replay that re-applies only its
233    /// head entry after a crash (a box being restored) must NOT: its blob frames are durable
234    /// one by one and the document that publishes them may be several frames on.
235    fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
236        self.put_blob(bytes)
237    }
238
239    /// The bytes at this address, or `None` if nothing was ever stored there.
240    fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>>;
241
242    /// Which version `name` is at, or `None` for a document that has never been
243    /// written.
244    ///
245    /// **`None` is a legitimate state and a fault is not.** A read that could
246    /// not distinguish them would report an unreadable identity table as *no
247    /// principals are registered*, which is the failure that opens a server up.
248    fn read_doc(&self, name: &str) -> Result<Option<BlobId>>;
249
250    /// **The compare-and-swap.** Move `name` from `expected` to `new`, and
251    /// append the new version to the audit trail, atomically.
252    ///
253    /// `Ok(false)` means the document was not at `expected` — another writer
254    /// committed in between — and the caller must re-read and re-apply. It is
255    /// deliberately not an error: losing a race is an ordinary outcome of a
256    /// contended document and a caller that retried on `Err` would also retry
257    /// on a broken disk.
258    fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool>;
259
260    /// Every version `name` has had, oldest first (the control plane's audit trail).
261    fn history(&self, name: &str) -> Result<Vec<DocVersion>>;
262}
263
264/// **The RAM copy of the control database**, taken by [`ControlDb::suspend`]
265/// the instant before `control.redb`'s file is closed and held until
266/// [`ControlDb::resume`] reopens it. Every current document pointer, every
267/// blob (current versions, their shards, and the history — the blobs table
268/// whole, because a shard is reached from a document by id and nothing
269/// short of the table says which ids a document reaches), and every sealed
270/// secret row. Reads are answered from it verbatim; writes are refused by
271/// name ([`Error::OffStripe`]).
272#[derive(Debug, Clone)]
273pub struct RamCopy {
274    /// Why the store is off the stripe — the growth's phase, in words.
275    pub why: String,
276    docs: std::collections::BTreeMap<String, BlobId>,
277    blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
278    secrets: std::collections::BTreeMap<String, Vec<u8>>,
279    /// When it was taken, unix ms.
280    pub taken_unix_ms: crate::model::UnixMillis,
281}
282
283impl RamCopy {
284    /// How many bytes the copy holds (blobs and secret rows).
285    pub fn bytes(&self) -> u64 {
286        self.blobs.values().map(|b| b.len() as u64).sum::<u64>()
287            + self.secrets.values().map(|b| b.len() as u64).sum::<u64>()
288    }
289    /// How many documents have a current version.
290    pub fn documents(&self) -> usize {
291        self.docs.len()
292    }
293    /// The current pointer of every document, name → blob id hex — what
294    /// [`ControlDb::resume`] compares the disk against.
295    pub fn pointers(&self) -> Vec<(String, String)> {
296        self.docs
297            .iter()
298            .map(|(k, v)| (k.clone(), v.to_string()))
299            .collect()
300    }
301    /// A secret's raw row (tag ‖ sealed bytes), for the secret store.
302    pub(crate) fn secret_row(&self, key: &str) -> Option<&[u8]> {
303        self.secrets.get(key).map(|v| v.as_slice())
304    }
305}
306
307/// **The control database's current state, read without taking it off the stripe**: every
308/// document pointer, every blob, every secret row (`None` = a tombstone; values are the SEALED
309/// bytes). What a twin snapshot carries and what a restored box re-seals.
310#[derive(Debug, Clone, Default)]
311pub struct ControlExport {
312    /// Document name → current blob.
313    pub docs: std::collections::BTreeMap<String, BlobId>,
314    /// Every blob, current and historical.
315    pub blobs: std::collections::BTreeMap<BlobId, Vec<u8>>,
316    /// Secret key → sealed value, or `None` for a tombstone.
317    pub secrets: std::collections::BTreeMap<String, Option<Vec<u8>>>,
318}
319
320impl ControlDb {
321    /// [`ControlExport`] of the live database (or of the RAM copy while suspended), in one
322    /// read transaction.
323    pub fn export(&self) -> Result<ControlExport> {
324        let raw = self.read(
325            "exporting the control database",
326            |txn| {
327                let mut docs = std::collections::BTreeMap::new();
328                let mut blobs = std::collections::BTreeMap::new();
329                let table = txn
330                    .open_table(DOCUMENTS)
331                    .map_err(|err| faulted("opening the documents table", err))?;
332                for entry in table
333                    .iter()
334                    .map_err(|err| faulted("scanning the documents table", err))?
335                {
336                    let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
337                    if let Some(id) = BlobId::from_slice(v.value()) {
338                        docs.insert(k.value().to_owned(), id);
339                    }
340                }
341                let table = txn
342                    .open_table(BLOBS)
343                    .map_err(|err| faulted("opening the blobs table", err))?;
344                for entry in table
345                    .iter()
346                    .map_err(|err| faulted("scanning the blobs table", err))?
347                {
348                    let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
349                    if let Some(id) = BlobId::from_slice(k.value()) {
350                        blobs.insert(id, v.value().to_vec());
351                    }
352                }
353                Ok((docs, blobs, crate::secrets::RedbSecretStore::rows(txn)?))
354            },
355            |ram| Ok((ram.docs.clone(), ram.blobs.clone(), ram.secrets.clone())),
356        )?;
357        let (docs, blobs, rows) = raw;
358        let mut secrets = std::collections::BTreeMap::new();
359        for (key, row) in rows {
360            let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &row)
361                .map_err(|err| Error::ControlStore(err.to_string()))?;
362            secrets.insert(key, value);
363        }
364        Ok(ControlExport {
365            docs,
366            blobs,
367            secrets,
368        })
369    }
370}
371
372/// One row of [`ControlDb::export_each`], borrowed from the read transaction (or the RAM copy):
373/// nothing of the database is gathered to hand it over.
374#[derive(Debug, Clone, Copy)]
375pub enum ExportRow<'a> {
376    /// A blob, current or historical.
377    Blob(BlobId, &'a [u8]),
378    /// A document's current pointer.
379    Doc(&'a str, BlobId),
380    /// A secret row: the sealed value, or `None` for a tombstone.
381    Secret(&'a str, Option<&'a [u8]>),
382}
383
384impl ControlDb {
385    /// **[`ControlDb::export`], row by row** — every blob, then every document pointer, then
386    /// every secret row, each handed to `row` as it is read, in one read transaction. What a
387    /// twin snapshot streams: the database never lies in memory whole. `row`'s error stops the
388    /// scan and comes back as `Ok(Err(e))`; the database's own faults are the outer error.
389    pub fn export_each<E>(
390        &self,
391        row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
392    ) -> Result<std::result::Result<(), E>> {
393        let row = std::cell::RefCell::new(row);
394        self.read(
395            "exporting the control database row by row",
396            |txn| export_txn(txn, &mut **row.borrow_mut()),
397            |ram| export_ram(ram, &mut **row.borrow_mut()),
398        )
399    }
400
401    /// **A consistent view of the whole database, taken NOW and read LATER** (R5.1-A,
402    /// 2026-10-07): a redb read transaction (MVCC — it holds the state of this instant while
403    /// writers go on), or the RAM copy while the store is off the stripe. What a snapshot takes
404    /// under the log's freeze, so it can stream the rows AFTER the freeze is gone.
405    ///
406    /// Said plainly: while a view lives, redb keeps the pages it reads and the file stays open —
407    /// a growth that takes the store off the stripe meanwhile cannot unmount until the view goes.
408    pub fn view(&self) -> Result<ControlView> {
409        if let Some(copy) = read_lock(&self.ram).as_ref() {
410            return Ok(ControlView(View::Ram(Box::new(copy.clone()))));
411        }
412        let guard = read_lock(&self.live);
413        let Some(db) = guard.as_ref() else {
414            return Err(self.off_stripe());
415        };
416        let txn = db
417            .begin_read()
418            .map_err(|err| faulted("beginning a view of the control database", err))?;
419        Ok(ControlView(View::Live(Box::new(txn))))
420    }
421}
422
423/// See [`ControlDb::view`].
424pub struct ControlView(View);
425
426enum View {
427    Live(Box<redb::ReadTransaction>),
428    Ram(Box<RamCopy>),
429}
430
431impl ControlView {
432    /// **Every last-seen row at the view's instant** ([`crate::activity`]):
433    /// what a snapshot carries. Empty for a database that never kept one and
434    /// for the RAM copy.
435    pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
436        match &self.0 {
437            View::Live(txn) => activity_rows(txn),
438            View::Ram(_) => Ok(Vec::new()),
439        }
440    }
441
442    /// [`ControlDb::export_each`] over the state the view was taken at.
443    pub fn export_each<E>(
444        &self,
445        row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
446    ) -> Result<std::result::Result<(), E>> {
447        match &self.0 {
448            View::Live(txn) => export_txn(txn, row),
449            View::Ram(ram) => export_ram(ram, row),
450        }
451    }
452}
453
454/// Every row of the last-seen table in `txn`; none when it was never made.
455fn activity_rows(txn: &redb::ReadTransaction) -> Result<Vec<(String, Vec<u8>)>> {
456    let table = match txn.open_table(PRINCIPAL_ACTIVITY) {
457        Ok(t) => t,
458        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
459        Err(err) => return Err(faulted("opening the last-seen table", err)),
460    };
461    let mut rows = Vec::new();
462    for entry in table
463        .iter()
464        .map_err(|err| faulted("scanning the last-seen table", err))?
465    {
466        let (k, v) = entry.map_err(|err| faulted("reading a last-seen row", err))?;
467        rows.push((k.value().to_owned(), v.value().to_vec()));
468    }
469    Ok(rows)
470}
471
472fn export_txn<E>(
473    txn: &redb::ReadTransaction,
474    row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
475) -> Result<std::result::Result<(), E>> {
476    let table = txn
477        .open_table(BLOBS)
478        .map_err(|err| faulted("opening the blobs table", err))?;
479    for entry in table
480        .iter()
481        .map_err(|err| faulted("scanning the blobs table", err))?
482    {
483        let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
484        if let Some(id) = BlobId::from_slice(k.value()) {
485            if let Err(e) = row(ExportRow::Blob(id, v.value())) {
486                return Ok(Err(e));
487            }
488        }
489    }
490    let table = txn
491        .open_table(DOCUMENTS)
492        .map_err(|err| faulted("opening the documents table", err))?;
493    for entry in table
494        .iter()
495        .map_err(|err| faulted("scanning the documents table", err))?
496    {
497        let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
498        if let Some(id) = BlobId::from_slice(v.value()) {
499            if let Err(e) = row(ExportRow::Doc(k.value(), id)) {
500                return Ok(Err(e));
501            }
502        }
503    }
504    for (key, raw) in crate::secrets::RedbSecretStore::rows(txn)? {
505        let value = crate::secrets::RedbSecretStore::decode_row_of(&key, &raw)
506            .map_err(|err| Error::ControlStore(err.to_string()))?;
507        if let Err(e) = row(ExportRow::Secret(&key, value.as_deref())) {
508            return Ok(Err(e));
509        }
510    }
511    Ok(Ok(()))
512}
513
514fn export_ram<E>(
515    ram: &RamCopy,
516    row: &mut dyn FnMut(ExportRow<'_>) -> std::result::Result<(), E>,
517) -> Result<std::result::Result<(), E>> {
518    for (id, bytes) in &ram.blobs {
519        if let Err(e) = row(ExportRow::Blob(*id, bytes)) {
520            return Ok(Err(e));
521        }
522    }
523    for (name, id) in &ram.docs {
524        if let Err(e) = row(ExportRow::Doc(name, *id)) {
525            return Ok(Err(e));
526        }
527    }
528    for (key, raw) in &ram.secrets {
529        let value = crate::secrets::RedbSecretStore::decode_row_of(key, raw)
530            .map_err(|err| Error::ControlStore(err.to_string()))?;
531        if let Err(e) = row(ExportRow::Secret(key, value.as_deref())) {
532            return Ok(Err(e));
533        }
534    }
535    Ok(Ok(()))
536}
537
538/// The most a RAM copy may weigh before [`ControlDb::suspend`] refuses:
539/// the control database is small by design (documents and shards, no git
540/// objects), and a copy past this is a sign something else is in it.
541pub const RAM_COPY_CEILING_BYTES: u64 = 512 << 20;
542
543/// What [`ControlDb::resume`] found when it compared the disk to the copy.
544#[derive(Debug, Clone, PartialEq, Eq)]
545pub struct ResumeReport {
546    /// Documents whose pointer on disk equals the pointer in RAM.
547    pub documents_matched: usize,
548    /// `name: ram <hex> vs disk <hex|absent>` for every document that
549    /// differs — empty is the expected answer, since every write was refused
550    /// while the copy stood in.
551    pub drift: Vec<String>,
552    /// How long the store was off the stripe, ms.
553    pub off_stripe_ms: i64,
554}
555
556/// **The one `redb::Database` under the control plane — and the door the
557/// runtime data-set growth takes it off the stripe through.**
558///
559/// [`RedbControlStore`] and [`RedbSecretStore`] both hold an `Arc` of this
560/// rather than of the database, so that ONE call can close the file under
561/// every handle in the process and one call can reopen it, with nothing
562/// upstream — `AuthTable`, `RepoCatalog`, the SSH `Control`, the gRPC gate
563/// — holding a stale descriptor that would pin the mount. That is the whole
564/// mechanism behind `DATA-SET-GROWTH-FLOW.md` G4/T2: `control.redb` lives
565/// on the stripe, every SSH authentication reads it at call time, and the
566/// stripe has to be unmounted for the members to be grown. So:
567///
568/// * [`ControlDb::suspend`] takes the write side (waiting for in-flight
569///   transactions), copies the tables into a [`RamCopy`], and drops the
570///   database — the file is closed and `umount /data` can succeed;
571/// * while suspended every read is served from the copy, byte for byte, so
572///   `principal_by_fingerprint`, the grant check and the bootstrap state keep
573///   answering; every write is [`Error::OffStripe`];
574/// * [`ControlDb::resume`] reopens the file, compares every document
575///   pointer on disk with the copy (they must agree — no write got through)
576///   and drops the copy.
577///
578/// An in-memory database (tests, `--dry-run`) has no file to close: it is
579/// kept, and `suspend` still switches every reader onto the copy and every
580/// writer onto the refusal, so the semantics are testable without a disk.
581pub struct ControlDb {
582    path: Option<std::path::PathBuf>,
583    live: std::sync::RwLock<Option<redb::Database>>,
584    ram: std::sync::RwLock<Option<RamCopy>>,
585    /// The group transaction open on one thread, if any ([`ControlDb::begin_group`]).
586    group: std::sync::Mutex<Option<Group>>,
587    /// Durable commits (each one fsync) since open ([`ControlDb::durable_commits`]).
588    durable_commits: std::sync::atomic::AtomicU64,
589}
590
591impl std::fmt::Debug for ControlDb {
592    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
593        f.debug_struct("ControlDb")
594            .field("path", &self.path)
595            .field("suspended", &self.is_suspended())
596            .finish()
597    }
598}
599
600fn read_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
601    lock.read().unwrap_or_else(|poison| poison.into_inner())
602}
603
604fn write_lock<T>(lock: &std::sync::RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
605    lock.write().unwrap_or_else(|poison| poison.into_inner())
606}
607
608impl ControlDb {
609    fn new(path: Option<std::path::PathBuf>, db: redb::Database) -> Self {
610        ControlDb {
611            path,
612            live: std::sync::RwLock::new(Some(db)),
613            ram: std::sync::RwLock::new(None),
614            group: std::sync::Mutex::new(None),
615            durable_commits: std::sync::atomic::AtomicU64::new(0),
616        }
617    }
618
619    /// A handle over an in-memory database somebody else prepared — the
620    /// secret store's own tests build one without a control store.
621    #[cfg(test)]
622    pub(crate) fn in_memory_for_tests(db: redb::Database) -> Self {
623        Self::new(None, db)
624    }
625
626    /// Is the store off the stripe right now?
627    pub fn is_suspended(&self) -> bool {
628        read_lock(&self.ram).is_some()
629    }
630
631    /// The RAM copy, while suspended.
632    pub fn ram_copy(&self) -> Option<RamCopy> {
633        read_lock(&self.ram).clone()
634    }
635
636    /// The refusal every write meets while suspended.
637    fn off_stripe(&self) -> Error {
638        Error::OffStripe {
639            why: read_lock(&self.ram)
640                .as_ref()
641                .map(|r| r.why.clone())
642                .unwrap_or_else(|| "suspended".to_owned()),
643        }
644    }
645
646    /// Run `f` over a read transaction on the live database, or, while
647    /// suspended, `ram` over the copy. The read guard is held for the
648    /// transaction's whole life, which is what makes `suspend` wait for it.
649    pub(crate) fn read<T>(
650        &self,
651        what: &str,
652        live: impl FnOnce(&redb::ReadTransaction) -> Result<T>,
653        ram: impl FnOnce(&RamCopy) -> Result<T>,
654    ) -> Result<T> {
655        if let Some(copy) = read_lock(&self.ram).as_ref() {
656            return ram(copy);
657        }
658        let guard = read_lock(&self.live);
659        let Some(db) = guard.as_ref() else {
660            return Err(self.off_stripe());
661        };
662        let txn = db
663            .begin_read()
664            .map_err(|err| faulted(&format!("beginning {what}"), err))?;
665        live(&txn)
666    }
667
668    /// Run `f` over a write transaction on the live database; refused while
669    /// suspended. `f` says whether to keep what it wrote ([`Txn`]); the commit is
670    /// made here — or, on the thread that holds a group ([`ControlDb::begin_group`]),
671    /// left to the group's one commit.
672    pub(crate) fn write<T>(
673        &self,
674        what: &str,
675        f: impl FnOnce(&redb::WriteTransaction) -> Result<Txn<T>>,
676    ) -> Result<T> {
677        if self.is_suspended() {
678            return Err(self.off_stripe());
679        }
680        {
681            let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
682            if let Some(g) = group
683                .as_mut()
684                .filter(|g| g.owner == std::thread::current().id())
685            {
686                return match f(&g.txn) {
687                    Ok(Txn::Commit(t) | Txn::CommitUnsynced(t)) => {
688                        g.dirty = true;
689                        Ok(t)
690                    }
691                    Ok(Txn::Discard(t)) => Ok(t),
692                    Err(err) => {
693                        g.poisoned = true;
694                        Err(err)
695                    }
696                };
697            }
698        }
699        let guard = read_lock(&self.live);
700        let Some(db) = guard.as_ref() else {
701            return Err(self.off_stripe());
702        };
703        let mut txn = db
704            .begin_write()
705            .map_err(|err| faulted(&format!("beginning {what}"), err))?;
706        match f(&txn)? {
707            Txn::CommitUnsynced(t) => {
708                txn.set_durability(redb::Durability::None);
709                txn.commit()
710                    .map_err(|err| faulted(&format!("committing {what}"), err))?;
711                Ok(t)
712            }
713            Txn::Commit(t) => {
714                txn.commit()
715                    .map_err(|err| faulted(&format!("committing {what}"), err))?;
716                self.durable_commits
717                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
718                Ok(t)
719            }
720            // Dropped, not committed: redb rolls it back.
721            Txn::Discard(t) => Ok(t),
722        }
723    }
724
725    /// **Open a group on this thread** (perf twin-apply-fsync, 2026-10-06): every write this
726    /// thread makes through the stores over this database goes into ONE transaction until
727    /// [`ControlDb::commit_group`] or [`ControlDb::abort_group`]. Writes from other threads wait
728    /// for it (redb takes one writer at a time); reads see the state before it. A twin's
729    /// follower holds one for the control-plane entries of each group it takes.
730    pub fn begin_group(&self) -> Result<()> {
731        if self.is_suspended() {
732            return Err(self.off_stripe());
733        }
734        let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
735        if group.is_some() {
736            return Err(Error::ControlStore(
737                "a group transaction is already open on this control store".to_owned(),
738            ));
739        }
740        let guard = read_lock(&self.live);
741        let Some(db) = guard.as_ref() else {
742            return Err(self.off_stripe());
743        };
744        let txn = db
745            .begin_write()
746            .map_err(|err| faulted("beginning a group of writes", err))?;
747        *group = Some(Group {
748            owner: std::thread::current().id(),
749            txn,
750            dirty: false,
751            poisoned: false,
752        });
753        Ok(())
754    }
755
756    /// **End the group with ONE durable commit, the marker `key` = `seq` inside it** — when it
757    /// wrote anything, or `mark` asks for the marker anyway. A group that wrote nothing and is not
758    /// asked to mark is dropped (no I/O). A group a write failed in is aborted whole and the
759    /// failure answered: none of it is kept. Answers whether it committed.
760    pub fn commit_group(&self, key: &str, seq: u64, mark: bool) -> Result<bool> {
761        let g = self.take_group()?;
762        if g.poisoned {
763            drop(g.txn);
764            return Err(Error::ControlStore(
765                "a write inside the group failed; the whole group was rolled back".to_owned(),
766            ));
767        }
768        if !g.dirty && !mark {
769            return Ok(false);
770        }
771        {
772            let mut marks = g
773                .txn
774                .open_table(MARKS)
775                .map_err(|err| faulted("opening the marks table", err))?;
776            marks
777                .insert(key, seq)
778                .map_err(|err| faulted("moving a marker", err))?;
779        }
780        g.txn
781            .commit()
782            .map_err(|err| faulted("committing a group of writes", err))?;
783        self.durable_commits
784            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
785        Ok(true)
786    }
787
788    /// **How many durable commits (one fsync each) this database made** since it was opened —
789    /// what the I/O tests hold to a bound.
790    pub fn durable_commits(&self) -> u64 {
791        self.durable_commits
792            .load(std::sync::atomic::Ordering::Relaxed)
793    }
794
795    /// Drop the group on this thread, keeping none of it.
796    pub fn abort_group(&self) {
797        let _ = self.take_group();
798    }
799
800    fn take_group(&self) -> Result<Group> {
801        let mut group = self.group.lock().unwrap_or_else(|p| p.into_inner());
802        match group.take() {
803            Some(g) if g.owner == std::thread::current().id() => Ok(g),
804            Some(g) => {
805                *group = Some(g);
806                Err(Error::ControlStore(
807                    "the open group transaction belongs to another thread".to_owned(),
808                ))
809            }
810            None => Err(Error::ControlStore(
811                "no group transaction is open".to_owned(),
812            )),
813        }
814    }
815
816    /// **Move the marker `key` to `seq`, durably** — inside this thread's group when one is open
817    /// (made durable with it), else in a commit of its own (one fsync, and every earlier
818    /// non-durable commit with it).
819    pub fn set_mark(&self, key: &str, seq: u64) -> Result<()> {
820        self.write("a marker", |txn| {
821            let mut marks = txn
822                .open_table(MARKS)
823                .map_err(|err| faulted("opening the marks table", err))?;
824            marks
825                .insert(key, seq)
826                .map_err(|err| faulted("moving a marker", err))?;
827            Ok(Txn::Commit(()))
828        })
829    }
830
831    /// The marker `key`, `None` when it was never set (a box from before markers, or a fresh one).
832    pub fn mark(&self, key: &str) -> Result<Option<u64>> {
833        self.read(
834            "a marker read",
835            |txn| {
836                let table = match txn.open_table(MARKS) {
837                    Ok(t) => t,
838                    Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
839                    Err(err) => return Err(faulted("opening the marks table", err)),
840                };
841                let found = table
842                    .get(key)
843                    .map_err(|err| faulted("reading a marker", err))?;
844                Ok(found.map(|v| v.value()))
845            },
846            |_ram| Ok(None),
847        )
848    }
849
850    /// **Every principal's last-seen row** ([`crate::activity`]): principal id →
851    /// its encoded stamps. Empty for a database that never kept one, and while
852    /// the store is off the stripe (the RAM copy carries no such table).
853    pub fn principal_activity(&self) -> Result<Vec<(String, Vec<u8>)>> {
854        self.read("a last-seen read", activity_rows, |_ram| Ok(Vec::new()))
855    }
856
857    /// **Write last-seen rows in ONE transaction** (one durable commit for the
858    /// whole batch), and the marker `mark` = 1 in the same transaction when one
859    /// is named. Refused while the store is off the stripe; the caller keeps
860    /// the rows and tries again.
861    pub fn put_principal_activity(
862        &self,
863        rows: &[(String, Vec<u8>)],
864        mark: Option<&str>,
865    ) -> Result<()> {
866        self.write("last-seen rows", |txn| {
867            {
868                let mut table = txn
869                    .open_table(PRINCIPAL_ACTIVITY)
870                    .map_err(|err| faulted("opening the last-seen table", err))?;
871                for (id, value) in rows {
872                    table
873                        .insert(id.as_str(), value.as_slice())
874                        .map_err(|err| faulted("writing a last-seen row", err))?;
875                }
876            }
877            if let Some(key) = mark {
878                let mut marks = txn
879                    .open_table(MARKS)
880                    .map_err(|err| faulted("opening the marks table", err))?;
881                marks
882                    .insert(key, 1u64)
883                    .map_err(|err| faulted("moving a marker", err))?;
884            }
885            Ok(Txn::Commit(()))
886        })
887    }
888
889    /// **Take the store off the stripe.** Copies every table into RAM, then
890    /// closes the file (an in-memory database is kept). Refuses when already
891    /// suspended, and when the copy would weigh more than
892    /// [`RAM_COPY_CEILING_BYTES`] — nothing is changed in that case.
893    pub fn suspend(&self, why: &str) -> Result<RamCopy> {
894        if self
895            .group
896            .lock()
897            .unwrap_or_else(|p| p.into_inner())
898            .is_some()
899        {
900            return Err(Error::ControlStore(
901                "a group of writes is open on the control store; it is suspended after the group ends".to_owned(),
902            ));
903        }
904        let mut live = write_lock(&self.live);
905        if self.is_suspended() {
906            return Err(Error::ControlStore(
907                "the control store is already suspended".to_owned(),
908            ));
909        }
910        let Some(db) = live.as_ref() else {
911            return Err(Error::ControlStore(
912                "the control store has no database to suspend".to_owned(),
913            ));
914        };
915        let txn = db
916            .begin_read()
917            .map_err(|err| faulted("beginning the suspend snapshot", err))?;
918        let mut copy = RamCopy {
919            why: why.to_owned(),
920            docs: Default::default(),
921            blobs: Default::default(),
922            secrets: Default::default(),
923            taken_unix_ms: crate::model::now_ms(),
924        };
925        {
926            let table = txn
927                .open_table(DOCUMENTS)
928                .map_err(|err| faulted("opening the documents table", err))?;
929            for entry in table
930                .iter()
931                .map_err(|err| faulted("scanning the documents table", err))?
932            {
933                let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
934                if let Some(id) = BlobId::from_slice(v.value()) {
935                    copy.docs.insert(k.value().to_owned(), id);
936                }
937            }
938            let table = txn
939                .open_table(BLOBS)
940                .map_err(|err| faulted("opening the blobs table", err))?;
941            for entry in table
942                .iter()
943                .map_err(|err| faulted("scanning the blobs table", err))?
944            {
945                let (k, v) = entry.map_err(|err| faulted("reading a blob", err))?;
946                if let Some(id) = BlobId::from_slice(k.value()) {
947                    copy.blobs.insert(id, v.value().to_vec());
948                }
949            }
950            copy.secrets = crate::secrets::RedbSecretStore::rows(&txn)?;
951        }
952        drop(txn);
953        let bytes = copy.bytes();
954        if bytes > RAM_COPY_CEILING_BYTES {
955            return Err(Error::ControlStore(format!(
956                "the control database would take {bytes} bytes of RAM to hold off the stripe, \
957                 over the {RAM_COPY_CEILING_BYTES}-byte ceiling; not suspending"
958            )));
959        }
960        // The copy is installed BEFORE the file is closed, so no reader ever
961        // meets a `None` database: the `ram` lock is taken after `live`'s
962        // write lock, which every reader takes (shared) first — no reader is
963        // inside the database at this point.
964        *write_lock(&self.ram) = Some(copy.clone());
965        if self.path.is_some() {
966            // Dropping the database closes the file: redb has no explicit
967            // close, and its `Drop` is what releases the descriptor and the
968            // lock. Every handle upstream is an `Arc<ControlDb>`, so nothing
969            // else holds the file.
970            *live = None;
971        }
972        Ok(copy)
973    }
974
975    /// **Put the store back on the stripe.** Reopens the file (the mount
976    /// must be back; a missing file is an error, never a fresh database),
977    /// compares every document pointer on disk with the copy, and drops the
978    /// copy. Refuses when not suspended.
979    pub fn resume(&self) -> Result<ResumeReport> {
980        let mut live = write_lock(&self.live);
981        let Some(copy) = read_lock(&self.ram).clone() else {
982            return Err(Error::ControlStore(
983                "the control store is not suspended".to_owned(),
984            ));
985        };
986        if live.is_none() {
987            let path = self.path.as_ref().ok_or_else(|| {
988                Error::ControlStore("no path to reopen the control store from".to_owned())
989            })?;
990            let db = redb::Database::open(path).map_err(|err| {
991                faulted(
992                    &format!("reopening the control store {}", path.display()),
993                    err,
994                )
995            })?;
996            *live = Some(db);
997        }
998        let db = live.as_ref().expect("installed above");
999        let txn = db
1000            .begin_read()
1001            .map_err(|err| faulted("beginning the resume check", err))?;
1002        let table = txn
1003            .open_table(DOCUMENTS)
1004            .map_err(|err| faulted("opening the documents table", err))?;
1005        let mut matched = 0;
1006        let mut drift = Vec::new();
1007        for (name, ram_id) in &copy.docs {
1008            let disk = table
1009                .get(name.as_str())
1010                .map_err(|err| faulted("reading a document pointer", err))?
1011                .and_then(|v| BlobId::from_slice(v.value()));
1012            match disk {
1013                Some(id) if id == *ram_id => matched += 1,
1014                other => drift.push(format!(
1015                    "{name}: ram {ram_id} vs disk {}",
1016                    other
1017                        .map(|i| i.to_string())
1018                        .unwrap_or_else(|| "absent".to_owned())
1019                )),
1020            }
1021        }
1022        for entry in table
1023            .iter()
1024            .map_err(|err| faulted("scanning the documents table", err))?
1025        {
1026            let (k, v) = entry.map_err(|err| faulted("reading a document pointer", err))?;
1027            if !copy.docs.contains_key(k.value()) {
1028                drift.push(format!(
1029                    "{}: ram absent vs disk {}",
1030                    k.value(),
1031                    BlobId::from_slice(v.value())
1032                        .map(|i| i.to_string())
1033                        .unwrap_or_else(|| "?".to_owned())
1034                ));
1035            }
1036        }
1037        drop(table);
1038        drop(txn);
1039        *write_lock(&self.ram) = None;
1040        Ok(ResumeReport {
1041            documents_matched: matched,
1042            drift,
1043            off_stripe_ms: crate::model::now_ms() - copy.taken_unix_ms,
1044        })
1045    }
1046}
1047
1048/// The redb-backed [`ControlStore`]: one file, three tables.
1049///
1050/// Cheap to clone; every clone is the same database. redb serialises write
1051/// transactions internally, so two threads committing at once is a wait and
1052/// never a lock error — which is why [`ControlStore::commit_doc`] has one
1053/// failure mode (the expectation did not hold) where the reference store had
1054/// two (that, and *somebody else holds the lock*).
1055#[derive(Clone)]
1056pub struct RedbControlStore {
1057    db: Arc<ControlDb>,
1058}
1059
1060impl std::fmt::Debug for RedbControlStore {
1061    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1062        f.write_str("RedbControlStore")
1063    }
1064}
1065
1066fn faulted(what: &str, err: impl std::fmt::Display) -> Error {
1067    Error::ControlStore(format!("{what}: {err}"))
1068}
1069
1070impl RedbControlStore {
1071    /// Open — creating if absent — the control database at `path`.
1072    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1073        let path = path.as_ref();
1074        let db = redb::Database::create(path).map_err(|err| {
1075            faulted(
1076                &format!("opening the control store {}", path.display()),
1077                err,
1078            )
1079        })?;
1080        Self::prepared(Some(path.to_path_buf()), db)
1081    }
1082
1083    /// **The handle the runtime growth suspends and resumes.** The same
1084    /// `Arc` the secret store holds.
1085    pub fn control_db(&self) -> Arc<ControlDb> {
1086        Arc::clone(&self.db)
1087    }
1088
1089    /// A control store with no file under it, for tests and for a `--dry-run`
1090    /// that must not touch the data path.
1091    pub fn in_memory() -> Result<Self> {
1092        let db = redb::Database::builder()
1093            .create_with_backend(redb::backends::InMemoryBackend::new())
1094            .map_err(|err| faulted("opening an in-memory control store", err))?;
1095        Self::prepared(None, db)
1096    }
1097
1098    /// Create every table up front.
1099    ///
1100    /// redb refuses to *read* a table that has never been written, so a store
1101    /// whose tables were created lazily would answer a first read with a fault
1102    /// rather than with "empty" — the exact confusion
1103    /// [`ControlStore::read_doc`] exists to prevent.
1104    fn prepared(path: Option<std::path::PathBuf>, db: redb::Database) -> Result<Self> {
1105        let txn = db
1106            .begin_write()
1107            .map_err(|err| faulted("beginning the control store's first transaction", err))?;
1108        {
1109            txn.open_table(DOCUMENTS)
1110                .map_err(|err| faulted("creating the documents table", err))?;
1111            txn.open_table(BLOBS)
1112                .map_err(|err| faulted("creating the blobs table", err))?;
1113            txn.open_table(DOC_HISTORY)
1114                .map_err(|err| faulted("creating the document history table", err))?;
1115            txn.open_table(MARKS)
1116                .map_err(|err| faulted("creating the marks table", err))?;
1117        }
1118        txn.commit()
1119            .map_err(|err| faulted("committing the control store's first transaction", err))?;
1120        crate::secrets::RedbSecretStore::prepare(&db)?;
1121        Ok(RedbControlStore {
1122            db: Arc::new(ControlDb::new(path, db)),
1123        })
1124    }
1125
1126    /// **The secret store in this same database** (2026-08-10).
1127    ///
1128    /// Handed out from here rather than opened separately, and that is the
1129    /// whole design: redb takes an exclusive file lock, so a second
1130    /// `redb::Database` over `control.redb` would fail. One file, one lock, one
1131    /// handle — and the secrets ride the transaction machinery the control
1132    /// plane already has instead of a second durability story beside it.
1133    ///
1134    /// It is **not** on [`ControlStore`]. That trait is about documents —
1135    /// `put_blob` / `commit_doc` / `history` — and its test decorators (the
1136    /// CAS-ignoring store `tests/control_cost.rs` proves the retry loop with,
1137    /// and the counting decorator) exist to break exactly those. A secret has
1138    /// none of that shape, so adding it to the trait would make every decorator
1139    /// carry a method it has no opinion about.
1140    pub fn secrets(&self) -> crate::secrets::RedbSecretStore {
1141        crate::secrets::RedbSecretStore::new(Arc::clone(&self.db))
1142    }
1143
1144    /// The next sequence number for `name`, read inside the caller's write
1145    /// transaction so two committers cannot pick the same one.
1146    fn next_seq(
1147        table: &impl redb::ReadableTable<(&'static str, u64), &'static [u8]>,
1148        name: &str,
1149    ) -> Result<u64> {
1150        let mut range = table
1151            .range((name, 0u64)..=(name, u64::MAX))
1152            .map_err(|err| faulted("reading a document's history", err))?;
1153        match range.next_back() {
1154            None => Ok(1),
1155            Some(entry) => {
1156                let (key, _) = entry.map_err(|err| faulted("reading a history row", err))?;
1157                Ok(key.value().1 + 1)
1158            }
1159        }
1160    }
1161}
1162
1163impl ControlStore for RedbControlStore {
1164    fn put_blob(&self, bytes: &[u8]) -> Result<BlobId> {
1165        let id = BlobId::of(bytes);
1166        self.db.write("a blob write", |txn| {
1167            {
1168                let mut table = txn
1169                    .open_table(BLOBS)
1170                    .map_err(|err| faulted("opening the blobs table", err))?;
1171                table
1172                    .insert(id.0.as_slice(), bytes)
1173                    .map_err(|err| faulted("storing a control blob", err))?;
1174            }
1175            Ok(Txn::Commit(id))
1176        })
1177    }
1178
1179    fn put_blob_unsynced(&self, bytes: &[u8]) -> Result<BlobId> {
1180        let id = BlobId::of(bytes);
1181        self.db.write("a blob write", |txn| {
1182            {
1183                let mut table = txn
1184                    .open_table(BLOBS)
1185                    .map_err(|err| faulted("opening the blobs table", err))?;
1186                table
1187                    .insert(id.0.as_slice(), bytes)
1188                    .map_err(|err| faulted("storing a control blob", err))?;
1189            }
1190            Ok(Txn::CommitUnsynced(id))
1191        })
1192    }
1193
1194    fn get_blob(&self, id: &BlobId) -> Result<Option<Vec<u8>>> {
1195        self.db.read(
1196            "a blob read",
1197            |txn| {
1198                let table = txn
1199                    .open_table(BLOBS)
1200                    .map_err(|err| faulted("opening the blobs table", err))?;
1201                let found = table
1202                    .get(id.0.as_slice())
1203                    .map_err(|err| faulted("reading a control blob", err))?;
1204                Ok(found.map(|value| value.value().to_vec()))
1205            },
1206            |ram| Ok(ram.blobs.get(id).cloned()),
1207        )
1208    }
1209
1210    fn read_doc(&self, name: &str) -> Result<Option<BlobId>> {
1211        self.db.read(
1212            "a document read",
1213            |txn| {
1214                let table = txn
1215                    .open_table(DOCUMENTS)
1216                    .map_err(|err| faulted("opening the documents table", err))?;
1217                let found = table
1218                    .get(name)
1219                    .map_err(|err| faulted("reading a control document pointer", err))?;
1220                match found {
1221                    None => Ok(None),
1222                    Some(value) => BlobId::from_slice(value.value())
1223                        .map(Some)
1224                        .ok_or_else(|| Error::CorruptDocument {
1225                            ref_name: name.to_owned(),
1226                            reason: format!(
1227                                "its stored pointer is {} bytes, not the {BLOB_ID_LEN} of a blob id",
1228                                value.value().len()
1229                            ),
1230                        }),
1231                }
1232            },
1233            |ram| Ok(ram.docs.get(name).copied()),
1234        )
1235    }
1236
1237    fn commit_doc(&self, name: &str, expected: Option<BlobId>, new: BlobId) -> Result<bool> {
1238        self.db.write("a control document commit", |txn| {
1239            {
1240                let mut documents = txn
1241                    .open_table(DOCUMENTS)
1242                    .map_err(|err| faulted("opening the documents table", err))?;
1243                let current = documents
1244                    .get(name)
1245                    .map_err(|err| faulted("reading a control document pointer", err))?
1246                    .and_then(|value| BlobId::from_slice(value.value()));
1247                if current != expected {
1248                    // Not an error: the caller re-reads and re-applies. Nothing
1249                    // was written yet, and the transaction is discarded.
1250                    return Ok(Txn::Discard(false));
1251                }
1252                documents
1253                    .insert(name, new.0.as_slice())
1254                    .map_err(|err| faulted("moving a control document pointer", err))?;
1255
1256                // The control plane's audit trail, in the SAME transaction as
1257                // the pointer move. Two transactions could be interrupted
1258                // between them, and a history that can disagree with the
1259                // document it describes is worse than no history at all.
1260                let mut history = txn
1261                    .open_table(DOC_HISTORY)
1262                    .map_err(|err| faulted("opening the document history table", err))?;
1263                let seq = Self::next_seq(&history, name)?;
1264                let mut row = Vec::with_capacity(8 + BLOB_ID_LEN);
1265                row.extend_from_slice(&crate::model::now_ms().to_be_bytes());
1266                row.extend_from_slice(&new.0);
1267                history
1268                    .insert((name, seq), row.as_slice())
1269                    .map_err(|err| faulted("appending to the document history", err))?;
1270            }
1271            Ok(Txn::Commit(true))
1272        })
1273    }
1274
1275    fn history(&self, name: &str) -> Result<Vec<DocVersion>> {
1276        self.db.read(
1277            "a history read",
1278            |txn| {
1279                let table = txn
1280                    .open_table(DOC_HISTORY)
1281                    .map_err(|err| faulted("opening the document history table", err))?;
1282                let range = table
1283                    .range((name, 0u64)..=(name, u64::MAX))
1284                    .map_err(|err| faulted("scanning a document's history", err))?;
1285                let mut out = Vec::new();
1286                for entry in range {
1287                    let (key, value) =
1288                        entry.map_err(|err| faulted("reading a history row", err))?;
1289                    let bytes = value.value();
1290                    if bytes.len() != 8 + BLOB_ID_LEN {
1291                        return Err(Error::CorruptDocument {
1292                            ref_name: name.to_owned(),
1293                            reason: format!(
1294                                "a history row is {} bytes, not {}",
1295                                bytes.len(),
1296                                8 + BLOB_ID_LEN
1297                            ),
1298                        });
1299                    }
1300                    let mut millis = [0u8; 8];
1301                    millis.copy_from_slice(&bytes[..8]);
1302                    out.push(DocVersion {
1303                        seq: key.value().1,
1304                        at: i64::from_be_bytes(millis),
1305                        blob: BlobId::from_slice(&bytes[8..]).expect("checked length"),
1306                    });
1307                }
1308                Ok(out)
1309            },
1310            // The trail is on the stripe and nothing wrote while it was off:
1311            // an audit read during the window is refused rather than answered
1312            // empty, which would read as "this document has no history".
1313            |ram| {
1314                Err(Error::OffStripe {
1315                    why: format!("{}; the audit trail is not held in RAM", ram.why),
1316                })
1317            },
1318        )
1319    }
1320}
1321
1322/// A control store, ready to be handed to [`crate::AuthTable`] and
1323/// [`crate::RepoCatalog`].
1324pub type SharedControlStore = Arc<dyn ControlStore>;
1325
1326/// The redb store at `path`, as a [`SharedControlStore`].
1327pub fn open_control_store(path: impl AsRef<Path>) -> Result<SharedControlStore> {
1328    Ok(Arc::new(RedbControlStore::open(path)?))
1329}
1330
1331/// **The whole control plane at `path`: the documents and the secrets, over one
1332/// database handle.**
1333///
1334/// The pair, not two calls, because the two halves must share the `redb`
1335/// handle: opening the file twice is an exclusive-lock failure, and this is the
1336/// only place that fact has to be known. `ServerStores::open` is the caller.
1337pub fn open_control(
1338    path: impl AsRef<Path>,
1339) -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
1340    let (store, secrets, _) = open_control_with_db(path)?;
1341    Ok((store, secrets))
1342}
1343
1344/// [`open_control`], and the [`ControlDb`] handle both halves share — what
1345/// the runtime data-set growth suspends and resumes (`gunnar-server`'s
1346/// `grow`). The handle is the same object either store reaches; it is
1347/// returned separately only because the store is type-erased on the way out.
1348pub fn open_control_with_db(
1349    path: impl AsRef<Path>,
1350) -> Result<(
1351    SharedControlStore,
1352    crate::secrets::RedbSecretStore,
1353    Arc<ControlDb>,
1354)> {
1355    let store = RedbControlStore::open(path)?;
1356    let secrets = store.secrets();
1357    let db = store.control_db();
1358    Ok((Arc::new(store), secrets, db))
1359}
1360
1361/// An in-memory control store, as a [`SharedControlStore`].
1362pub fn memory_control_store() -> Result<SharedControlStore> {
1363    Ok(Arc::new(RedbControlStore::in_memory()?))
1364}
1365
1366/// [`open_control`] with no file under it, for tests and for a `--dry-run` that
1367/// must not touch the data path.
1368pub fn memory_control() -> Result<(SharedControlStore, crate::secrets::RedbSecretStore)> {
1369    let store = RedbControlStore::in_memory()?;
1370    let secrets = store.secrets();
1371    Ok((Arc::new(store), secrets))
1372}
1373
1374#[cfg(test)]
1375mod tests {
1376    use super::*;
1377
1378    #[test]
1379    fn a_blob_id_is_the_hash_of_the_bytes_and_round_trips_through_hex() {
1380        let id = BlobId::of(b"the identity table");
1381        assert_eq!(BlobId::from_hex(&id.to_string()), Some(id));
1382        assert_ne!(id, BlobId::of(b"the identity tabl3"));
1383        assert_eq!(BlobId::from_hex("not hex"), None);
1384        assert_eq!(BlobId::from_hex(""), None);
1385    }
1386
1387    #[test]
1388    fn an_unwritten_document_reads_as_absent_and_a_written_one_reads_back() {
1389        let store = RedbControlStore::in_memory().expect("open");
1390        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), None);
1391        let id = store.put_blob(b"{}").expect("put");
1392        assert_eq!(
1393            store.get_blob(&id).expect("get").as_deref(),
1394            Some(&b"{}"[..])
1395        );
1396        assert!(store
1397            .commit_doc("refs/gunnar/auth", None, id)
1398            .expect("commit"));
1399        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
1400    }
1401
1402    /// **T2 — the store off the stripe** (`DATA-SET-GROWTH-FLOW.md` G4):
1403    /// while suspended every read answers from the RAM copy byte for byte,
1404    /// every write is refused by name with `OffStripe`, the audit trail is
1405    /// refused rather than answered empty, and a resume finds no drift and
1406    /// takes writes again. Suspending twice and resuming an unsuspended
1407    /// store are both refusals.
1408    #[test]
1409    fn a_suspended_store_reads_from_ram_refuses_writes_and_resumes_without_drift() {
1410        use gunnar_store::SecretStore as _;
1411        let store = RedbControlStore::in_memory().expect("open");
1412        let secrets = store.secrets();
1413        let db = store.control_db();
1414        let id = store.put_blob(b"{\"principals\":1}").expect("put");
1415        assert!(store
1416            .commit_doc("refs/gunnar/auth", None, id)
1417            .expect("commit"));
1418        secrets
1419            .put("host-key", b"sealed bytes")
1420            .expect("put secret");
1421        assert!(!db.is_suspended());
1422
1423        let copy = db.suspend("draining for a growth").expect("suspend");
1424        assert!(db.is_suspended());
1425        assert_eq!(copy.documents(), 1);
1426        assert!(copy.bytes() > 0);
1427        assert_eq!(copy.why, "draining for a growth");
1428        // Reads: served from the copy.
1429        assert_eq!(store.read_doc("refs/gunnar/auth").expect("read"), Some(id));
1430        assert_eq!(
1431            store.get_blob(&id).expect("get").as_deref(),
1432            Some(&b"{\"principals\":1}"[..])
1433        );
1434        assert_eq!(store.read_doc("never-written").expect("read"), None);
1435        assert_eq!(
1436            secrets.get("host-key").expect("get").as_deref(),
1437            Some(&b"sealed bytes"[..])
1438        );
1439        assert_eq!(secrets.get("absent").expect("get"), None);
1440        assert_eq!(
1441            secrets
1442                .get_or_create("host-key", &mut || panic!("must not mint"))
1443                .expect("held"),
1444            b"sealed bytes"
1445        );
1446        // Writes: refused, by name, with the phase in the words.
1447        let refused = store.put_blob(b"new").unwrap_err();
1448        assert!(matches!(refused, Error::OffStripe { .. }), "{refused:?}");
1449        assert!(
1450            refused.to_string().contains("draining for a growth"),
1451            "{refused}"
1452        );
1453        assert!(refused.to_string().contains("retry"), "{refused}");
1454        let other = BlobId::of(b"other");
1455        assert!(matches!(
1456            store
1457                .commit_doc("refs/gunnar/auth", Some(id), other)
1458                .unwrap_err(),
1459            Error::OffStripe { .. }
1460        ));
1461        assert!(matches!(
1462            store.history("refs/gunnar/auth").unwrap_err(),
1463            Error::OffStripe { .. }
1464        ));
1465        let sec = secrets.put("host-key", b"x").unwrap_err().to_string();
1466        assert!(sec.contains("off the stripe"), "{sec}");
1467        let mint = secrets
1468            .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
1469            .unwrap_err()
1470            .to_string();
1471        assert!(mint.contains("will not be minted"), "{mint}");
1472        assert!(
1473            db.suspend("again").is_err(),
1474            "a second suspend is a refusal"
1475        );
1476
1477        let report = db.resume().expect("resume");
1478        assert!(!db.is_suspended());
1479        assert_eq!(report.documents_matched, 1);
1480        assert!(report.drift.is_empty(), "{:?}", report.drift);
1481        assert!(
1482            db.resume().is_err(),
1483            "resuming an unsuspended store is a refusal"
1484        );
1485        // Back on the stripe: writes land, the trail is readable again.
1486        assert!(store
1487            .commit_doc("refs/gunnar/auth", Some(id), other)
1488            .expect("commit"));
1489        assert_eq!(store.history("refs/gunnar/auth").expect("history").len(), 2);
1490        assert_eq!(
1491            secrets
1492                .get_or_create("fresh", &mut || Ok(b"minted".to_vec()))
1493                .expect("mint"),
1494            b"minted"
1495        );
1496    }
1497
1498    /// **The file really closes.** On the stripe the whole point of the
1499    /// suspend is that `umount /data` can succeed, which needs the descriptor
1500    /// gone: `/proc/self/fd` names the file before, not during, and again
1501    /// after. Linux only, like the mount it exists for.
1502    #[cfg(target_os = "linux")]
1503    #[test]
1504    fn a_suspended_file_store_holds_no_descriptor_on_its_file() {
1505        let dir = tempfile::tempdir().expect("tempdir");
1506        let path = dir.path().join("control.redb");
1507        let store = RedbControlStore::open(&path).expect("open");
1508        let db = store.control_db();
1509        let id = store.put_blob(b"doc").expect("put");
1510        assert!(store.commit_doc("d", None, id).expect("commit"));
1511        let holds = |p: &Path| -> bool {
1512            std::fs::read_dir("/proc/self/fd")
1513                .expect("procfs")
1514                .flatten()
1515                .filter_map(|e| std::fs::read_link(e.path()).ok())
1516                .any(|t| t == p)
1517        };
1518        assert!(holds(&path), "the open store holds its file");
1519        db.suspend("unmounting").expect("suspend");
1520        assert!(
1521            !holds(&path),
1522            "a suspended store holds NO descriptor on its file"
1523        );
1524        assert_eq!(store.read_doc("d").expect("read from ram"), Some(id));
1525        db.resume().expect("resume");
1526        assert!(holds(&path), "a resumed store holds its file again");
1527        assert_eq!(store.read_doc("d").expect("read from disk"), Some(id));
1528    }
1529
1530    /// The compare-and-swap, seen to refuse. Without this the retry loop in
1531    /// `doc.rs` is decoration: a `commit_doc` that always returned `true` would
1532    /// satisfy every other test in this module.
1533    #[test]
1534    fn a_commit_against_the_wrong_expectation_is_refused_and_changes_nothing() {
1535        let store = RedbControlStore::in_memory().expect("open");
1536        let first = store.put_blob(b"one").expect("put");
1537        let second = store.put_blob(b"two").expect("put");
1538        assert!(store.commit_doc("doc", None, first).expect("commit"));
1539
1540        // Somebody else already moved it: this writer's expectation is stale.
1541        assert!(
1542            !store.commit_doc("doc", None, second).expect("commit"),
1543            "a commit that expected an absent document was accepted against a written one"
1544        );
1545        assert_eq!(
1546            store.read_doc("doc").expect("read"),
1547            Some(first),
1548            "a refused commit still moved the pointer"
1549        );
1550        assert_eq!(
1551            store.history("doc").expect("history").len(),
1552            1,
1553            "a refused commit still appended to the audit trail"
1554        );
1555
1556        // …and with the right expectation it lands.
1557        assert!(store
1558            .commit_doc("doc", Some(first), second)
1559            .expect("commit"));
1560        assert_eq!(store.read_doc("doc").expect("read"), Some(second));
1561    }
1562
1563    /// The control plane's audit trail: every version, in order, and it is per-document.
1564    #[test]
1565    fn the_history_is_append_only_and_one_document_does_not_see_anothers() {
1566        let store = RedbControlStore::in_memory().expect("open");
1567        let mut previous = None;
1568        let mut expected = Vec::new();
1569        for text in [b"a".as_slice(), b"bb", b"ccc"] {
1570            let id = store.put_blob(text).expect("put");
1571            assert!(store.commit_doc("auth", previous, id).expect("commit"));
1572            previous = Some(id);
1573            expected.push(id);
1574        }
1575        let other = store.put_blob(b"catalog").expect("put");
1576        assert!(store.commit_doc("catalog", None, other).expect("commit"));
1577
1578        let history = store.history("auth").expect("history");
1579        assert_eq!(
1580            history.iter().map(|v| v.blob).collect::<Vec<_>>(),
1581            expected,
1582            "the audit trail must list every version of the document, oldest first"
1583        );
1584        assert_eq!(
1585            history.iter().map(|v| v.seq).collect::<Vec<_>>(),
1586            vec![1, 2, 3]
1587        );
1588        assert_eq!(
1589            store.history("catalog").expect("history").len(),
1590            1,
1591            "one document's history picked up another document's versions"
1592        );
1593        assert_eq!(store.history("never-written").expect("history").len(), 0);
1594
1595        // Every historical version is still readable, which is what makes this
1596        // an audit trail rather than a list of hashes.
1597        assert_eq!(
1598            store.get_blob(&history[0].blob).expect("get").as_deref(),
1599            Some(&b"a"[..])
1600        );
1601    }
1602}