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` (sidecar), `next_segment_no`). CATALOG is rewritten on
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/// Persistent definition for a physical materialized-view table.
50#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
51pub struct MaterializedViewEntry {
52    pub name: String,
53    pub query: String,
54    pub last_refresh_epoch: u64,
55    #[serde(default)]
56    pub incremental: Option<IncrementalAggregateView>,
57}
58
59#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
60pub struct IncrementalAggregateView {
61    pub source_table: String,
62    pub source_table_id: u64,
63    pub group_column: u16,
64    pub group_output_column: u16,
65    pub outputs: Vec<IncrementalAggregateOutput>,
66    pub count_output_column: u16,
67    pub checkpoint_event_id: String,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
71pub struct IncrementalAggregateOutput {
72    pub output_column: u16,
73    pub kind: IncrementalAggregateKind,
74}
75
76#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
77pub enum IncrementalAggregateKind {
78    Count,
79    Sum { source_column: u16 },
80}
81
82/// The full in-memory catalog, mirrored on disk by [`write_atomic`].
83///
84/// Note: `open_generation` is intentionally **not** stored here — it bumps on
85/// every open, and keeping it in CATALOG would dirty the working tree even for
86/// a bare read. It lives in a separate sidecar file (`_meta/generation`) that
87/// callers can `.gitignore` for content-addressed storage workflows.
88#[derive(Debug, Clone, Serialize, Deserialize, Default)]
89pub struct Catalog {
90    /// Highest epoch ever assigned by this DB's commit sequencer.
91    pub db_epoch: u64,
92    /// Next table id to allocate.
93    pub next_table_id: u64,
94    /// Next shared-WAL segment number to allocate.
95    pub next_segment_no: u64,
96    pub tables: Vec<CatalogEntry>,
97    #[serde(default)]
98    pub procedures: Vec<ProcedureEntry>,
99    #[serde(default)]
100    pub triggers: Vec<TriggerEntry>,
101    #[serde(default)]
102    pub external_tables: Vec<ExternalTableEntry>,
103    #[serde(default)]
104    pub materialized_views: Vec<MaterializedViewEntry>,
105    #[serde(default)]
106    pub security: crate::security::SecurityCatalog,
107    /// Monotonic version for optimistic authorization snapshots.
108    #[serde(default)]
109    pub security_version: u64,
110    /// Catalog-level user accounts (Argon2id-hashed credentials).
111    #[serde(default)]
112    pub users: Vec<crate::auth::UserEntry>,
113    /// Catalog-level role definitions.
114    #[serde(default)]
115    pub roles: Vec<crate::auth::RoleEntry>,
116    /// Next monotonic user id to allocate.
117    #[serde(default)]
118    pub next_user_id: u64,
119    /// When true, every Database/Table/Transaction/MongrelSession operation
120    /// requires an authenticated `Principal` with sufficient permission.
121    /// Defaults to false → existing credentialless databases open unchanged.
122    /// See `docs/15-credential-enforcement.md`.
123    #[serde(default)]
124    pub require_auth: bool,
125}
126
127/// The `open_generation` counter, stored in `_meta/generation` (NOT in CATALOG).
128/// Bumped on every open to scope `txn_id` across reopens so ids never alias.
129///
130/// Kept as a sidecar so that CATALOG is stable across bare opens — the only
131/// volatile bytes in the database directory during a read-only session are
132/// this 8-byte file + `.lock` + caches, all of which can be `.gitignore`-d.
133pub const GENERATION_FILENAME: &str = "generation";
134
135/// Read `open_generation` from `_meta/generation`. Returns 0 if the file is
136/// missing (first open or after a fresh `create`). On legacy databases that
137/// stored `open_generation` inside CATALOG, the migration path is: if the
138/// sidecar doesn't exist, default to 0 (the counter only needs to be
139/// monotonic within a process's lifetime for txn-id uniqueness).
140pub fn read_generation(meta_dir: &Path) -> u64 {
141    let p = meta_dir.join(GENERATION_FILENAME);
142    match std::fs::read(&p) {
143        Ok(bytes) => {
144            if bytes.len() >= 8 {
145                let mut arr = [0u8; 8];
146                arr.copy_from_slice(&bytes[..8]);
147                u64::from_le_bytes(arr)
148            } else {
149                0
150            }
151        }
152        Err(_) => 0,
153    }
154}
155
156/// Write `open_generation` to `_meta/generation` atomically (temp + rename +
157/// fsync). This is intentionally a separate file from CATALOG so that CATALOG
158/// stays byte-stable across bare opens (the generation counter is the only
159/// field that changes on every open).
160pub fn write_generation(meta_dir: &Path, generation: u64) -> Result<()> {
161    std::fs::create_dir_all(meta_dir)?;
162    let tmp = meta_dir.join(format!(".{GENERATION_FILENAME}.tmp"));
163    {
164        let mut f = std::fs::File::create(&tmp)?;
165        f.write_all(&generation.to_le_bytes())?;
166        f.sync_all()?;
167    }
168    let dest = meta_dir.join(GENERATION_FILENAME);
169    std::fs::rename(&tmp, &dest)?;
170    if let Ok(d) = std::fs::File::open(meta_dir) {
171        let _ = d.sync_all();
172    }
173    Ok(())
174}
175
176impl Catalog {
177    /// An empty catalog for a freshly created DB.
178    pub fn empty() -> Self {
179        Catalog::default()
180    }
181
182    /// Look up an entry by name (live only).
183    pub fn live(&self, name: &str) -> Option<&CatalogEntry> {
184        self.tables
185            .iter()
186            .find(|t| t.name == name && matches!(t.state, TableState::Live))
187    }
188}
189
190#[cfg(feature = "encryption")]
191fn seal(body: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Vec<u8>> {
192    match meta_dek {
193        Some(dek) => crate::encryption::encrypt_blob(dek, body),
194        None => Ok(plaintext_frame(body)),
195    }
196}
197
198#[cfg(not(feature = "encryption"))]
199fn seal(body: &[u8], _meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Vec<u8>> {
200    Ok(plaintext_frame(body))
201}
202
203fn plaintext_frame(body: &[u8]) -> Vec<u8> {
204    let hash = Sha256::digest(body);
205    let mut out = Vec::with_capacity(body.len() + 8 + 32);
206    out.extend_from_slice(MAGIC);
207    out.extend_from_slice(&hash);
208    out.extend_from_slice(body);
209    out
210}
211
212/// Atomically write the catalog to `<dir>/CATALOG` (review fix #19: dir-fsync).
213///
214/// If `meta_dek` is `Some`, the body is AES-256-GCM sealed (confidential +
215/// authenticated); otherwise the body carries a SHA-256 tag (integrity only).
216pub fn write_atomic(
217    dir: &Path,
218    cat: &Catalog,
219    meta_dek: Option<&[u8; META_DEK_LEN]>,
220) -> Result<()> {
221    let body = serde_json::to_vec(cat)
222        .map_err(|e| MongrelError::Other(format!("catalog serialize: {e}")))?;
223    let payload = seal(&body, meta_dek)?;
224
225    let tmp = dir.join(format!(".{CATALOG_FILENAME}.tmp"));
226    {
227        let mut f = std::fs::File::create(&tmp)?;
228        f.write_all(&payload)?;
229        f.sync_all()?;
230    }
231    let dest = dir.join(CATALOG_FILENAME);
232    std::fs::rename(&tmp, &dest)?;
233    // Directory fsync so the rename is durable across a crash.
234    if let Ok(d) = std::fs::File::open(dir) {
235        let _ = d.sync_all();
236    }
237    Ok(())
238}
239
240#[cfg(feature = "encryption")]
241fn open_payload(bytes: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
242    match meta_dek {
243        Some(dek) => match crate::encryption::decrypt_blob(dek, bytes) {
244            Ok(body) => deserialize(&body),
245            Err(_) => Ok(None),
246        },
247        None => parse_plaintext(bytes),
248    }
249}
250
251#[cfg(not(feature = "encryption"))]
252fn open_payload(bytes: &[u8], _meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
253    parse_plaintext(bytes)
254}
255
256fn deserialize(body: &[u8]) -> Result<Option<Catalog>> {
257    serde_json::from_slice(body)
258        .map(Some)
259        .map_err(|e| MongrelError::Other(format!("catalog deserialize: {e}")))
260}
261
262/// Read the catalog from `<dir>/CATALOG`. Returns `Ok(None)` if no catalog is
263/// present, or if authentication fails (tampered / wrong key).
264pub fn read(dir: &Path, meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
265    let p = dir.join(CATALOG_FILENAME);
266    let bytes = match std::fs::read(&p) {
267        Ok(b) => b,
268        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
269        Err(e) => return Err(e.into()),
270    };
271    open_payload(&bytes, meta_dek)
272}
273
274fn parse_plaintext(bytes: &[u8]) -> Result<Option<Catalog>> {
275    if bytes.len() < 8 + 32 || &bytes[..8] != MAGIC {
276        return Ok(None);
277    }
278    let (tag, body) = bytes[8..].split_at(32);
279    let calc = Sha256::digest(body);
280    if tag != calc.as_slice() {
281        // tampered
282        return Ok(None);
283    }
284    deserialize(body)
285}
286
287#[cfg(test)]
288mod tests {
289    use super::*;
290
291    #[test]
292    fn empty_catalog_default() {
293        let c = Catalog::empty();
294        assert_eq!(c.db_epoch, 0);
295        assert_eq!(c.next_table_id, 0);
296        assert!(c.tables.is_empty());
297    }
298}
mongreldb_core/catalog.rs

mongreldb_core/
catalog.rs
