mongreldb_core/
catalog.rs

1//! DB-wide catalog checkpoint (spec §5.1).
2//!
3//! The catalog records every table's id, name, schema, and live/dropped state
4//! plus the DB-wide monotonic counters (`db_epoch`, `next_table_id`,
5//! `open_generation`, `next_segment_no`). It is rewritten atomically on every
6//! DDL and persisted to `<root>/CATALOG` with:
7//!
8//! - a fixed magic + SHA-256 integrity tag for plaintext, or
9//! - AES-256-GCM (`meta_dek`-derived via [`Kek::derive_meta_key`]) which both
10//!   encrypts and authenticates, plus
11//! - a directory `sync_all` after the atomic rename (review fix #19), so a crash
12//!   never leaves a half-linked catalog entry.
13
14use crate::error::{MongrelError, Result};
15use crate::external_table::ExternalTableEntry;
16use crate::procedure::ProcedureEntry;
17use crate::schema::Schema;
18use crate::trigger::TriggerEntry;
19use serde::{Deserialize, Serialize};
20use sha2::{Digest, Sha256};
21use std::io::Write;
22use std::path::Path;
23
24pub const CATALOG_FILENAME: &str = "CATALOG";
25const MAGIC: &[u8; 8] = b"MONGRCAT";
26/// 32-byte meta DEK length (matches [`crate::encryption::DEK_LEN`]).
27pub const META_DEK_LEN: usize = 32;
28
29/// Lifecycle state of a catalog table entry.
30#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
31pub enum TableState {
32    /// Live and queryable.
33    Live,
34    /// Logically dropped at `at_epoch`; the physical subdir is reaped once no
35    /// reader pins a snapshot older than `at_epoch`.
36    Dropped { at_epoch: u64 },
37}
38
39/// One row of the catalog.
40#[derive(Debug, Clone, Serialize, Deserialize)]
41pub struct CatalogEntry {
42    pub table_id: u64,
43    pub name: String,
44    pub schema: Schema,
45    pub state: TableState,
46    pub created_epoch: u64,
47}
48
49/// The full in-memory catalog, mirrored on disk by [`write_atomic`].
50#[derive(Debug, Clone, Serialize, Deserialize, Default)]
51pub struct Catalog {
52    /// Highest epoch ever assigned by this DB's commit sequencer.
53    pub db_epoch: u64,
54    /// Next table id to allocate.
55    pub next_table_id: u64,
56    /// Bumped (and fsynced) on every open to scope `txn_id` across reopens.
57    pub open_generation: u64,
58    /// Next shared-WAL segment number to allocate.
59    pub next_segment_no: u64,
60    pub tables: Vec<CatalogEntry>,
61    #[serde(default)]
62    pub procedures: Vec<ProcedureEntry>,
63    #[serde(default)]
64    pub triggers: Vec<TriggerEntry>,
65    #[serde(default)]
66    pub external_tables: Vec<ExternalTableEntry>,
67    /// Catalog-level user accounts (Argon2id-hashed credentials).
68    #[serde(default)]
69    pub users: Vec<crate::auth::UserEntry>,
70    /// Catalog-level role definitions.
71    #[serde(default)]
72    pub roles: Vec<crate::auth::RoleEntry>,
73    /// Next monotonic user id to allocate.
74    #[serde(default)]
75    pub next_user_id: u64,
76    /// When true, every Database/Table/Transaction/MongrelSession operation
77    /// requires an authenticated `Principal` with sufficient permission.
78    /// Defaults to false → existing credentialless databases open unchanged.
79    /// See `docs/15-credential-enforcement.md`.
80    #[serde(default)]
81    pub require_auth: bool,
82}
83
84impl Catalog {
85    /// An empty catalog for a freshly created DB.
86    pub fn empty() -> Self {
87        Catalog::default()
88    }
89
90    /// Look up an entry by name (live only).
91    pub fn live(&self, name: &str) -> Option<&CatalogEntry> {
92        self.tables
93            .iter()
94            .find(|t| t.name == name && matches!(t.state, TableState::Live))
95    }
96}
97
98#[cfg(feature = "encryption")]
99fn seal(body: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Vec<u8>> {
100    match meta_dek {
101        Some(dek) => crate::encryption::encrypt_blob(dek, body),
102        None => Ok(plaintext_frame(body)),
103    }
104}
105
106#[cfg(not(feature = "encryption"))]
107fn seal(body: &[u8], _meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Vec<u8>> {
108    Ok(plaintext_frame(body))
109}
110
111fn plaintext_frame(body: &[u8]) -> Vec<u8> {
112    let hash = Sha256::digest(body);
113    let mut out = Vec::with_capacity(body.len() + 8 + 32);
114    out.extend_from_slice(MAGIC);
115    out.extend_from_slice(&hash);
116    out.extend_from_slice(body);
117    out
118}
119
120/// Atomically write the catalog to `<dir>/CATALOG` (review fix #19: dir-fsync).
121///
122/// If `meta_dek` is `Some`, the body is AES-256-GCM sealed (confidential +
123/// authenticated); otherwise the body carries a SHA-256 tag (integrity only).
124pub fn write_atomic(
125    dir: &Path,
126    cat: &Catalog,
127    meta_dek: Option<&[u8; META_DEK_LEN]>,
128) -> Result<()> {
129    let body = serde_json::to_vec(cat)
130        .map_err(|e| MongrelError::Other(format!("catalog serialize: {e}")))?;
131    let payload = seal(&body, meta_dek)?;
132
133    let tmp = dir.join(format!(".{CATALOG_FILENAME}.tmp"));
134    {
135        let mut f = std::fs::File::create(&tmp)?;
136        f.write_all(&payload)?;
137        f.sync_all()?;
138    }
139    let dest = dir.join(CATALOG_FILENAME);
140    std::fs::rename(&tmp, &dest)?;
141    // Directory fsync so the rename is durable across a crash.
142    if let Ok(d) = std::fs::File::open(dir) {
143        let _ = d.sync_all();
144    }
145    Ok(())
146}
147
148#[cfg(feature = "encryption")]
149fn open_payload(bytes: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
150    match meta_dek {
151        Some(dek) => match crate::encryption::decrypt_blob(dek, bytes) {
152            Ok(body) => deserialize(&body),
153            Err(_) => Ok(None),
154        },
155        None => parse_plaintext(bytes),
156    }
157}
158
159#[cfg(not(feature = "encryption"))]
160fn open_payload(bytes: &[u8], _meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
161    parse_plaintext(bytes)
162}
163
164fn deserialize(body: &[u8]) -> Result<Option<Catalog>> {
165    serde_json::from_slice(body)
166        .map(Some)
167        .map_err(|e| MongrelError::Other(format!("catalog deserialize: {e}")))
168}
169
170/// Read the catalog from `<dir>/CATALOG`. Returns `Ok(None)` if no catalog is
171/// present, or if authentication fails (tampered / wrong key).
172pub fn read(dir: &Path, meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
173    let p = dir.join(CATALOG_FILENAME);
174    let bytes = match std::fs::read(&p) {
175        Ok(b) => b,
176        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
177        Err(e) => return Err(e.into()),
178    };
179    open_payload(&bytes, meta_dek)
180}
181
182fn parse_plaintext(bytes: &[u8]) -> Result<Option<Catalog>> {
183    if bytes.len() < 8 + 32 || &bytes[..8] != MAGIC {
184        return Ok(None);
185    }
186    let (tag, body) = bytes[8..].split_at(32);
187    let calc = Sha256::digest(body);
188    if tag != calc.as_slice() {
189        // tampered
190        return Ok(None);
191    }
192    deserialize(body)
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198
199    #[test]
200    fn empty_catalog_default() {
201        let c = Catalog::empty();
202        assert_eq!(c.db_epoch, 0);
203        assert_eq!(c.next_table_id, 0);
204        assert!(c.tables.is_empty());
205    }
206}
mongreldb_core/catalog.rs

mongreldb_core/
catalog.rs
