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//!
14//! # Versioned catalog commands (spec §10.6, S1F-001)
15//!
16//! CATALOG is demoted from sole authority to a checkpoint of the versioned
17//! catalog command state machine (see [`crate::catalog_cmds`]). `Catalog`
18//! carries `catalog_version` plus a bounded retained tail of applied
19//! [`crate::catalog_cmds::CatalogCommandRecord`]s (`command_log`), so the
20//! records ride inside the existing persistence mechanics with no on-disk
21//! format change: both this checkpoint and the `DdlOp::CatalogSnapshot` WAL
22//! payload (`DdlOp::encode_catalog`) serialize the whole `Catalog`, command
23//! history included. All four state-machine fields default to empty on decode
24//! and are omitted from serialization while empty, so pre-S1F-001 databases
25//! open byte-identically unchanged.
26//!
27//! New code mutates the catalog through [`Catalog::apply_command`], which
28//! validates the record, applies it deterministically, bumps
29//! `catalog_version`, and appends it to the bounded history;
30//! [`Catalog::apply_command_and_checkpoint`] additionally rewrites this
31//! checkpoint. `Database` mutation entry points authorize via `require` /
32//! `require_for` (see [`crate::catalog_cmds::required_permission`]) and then
33//! either apply a `CatalogCommand` or update catalog fields with the same
34//! checkpoint path — pure apply stays principal-free for deterministic
35//! replica replay.
36
37use crate::error::{MongrelError, Result};
38use crate::external_table::ExternalTableEntry;
39use crate::procedure::ProcedureEntry;
40use crate::schema::Schema;
41use crate::trigger::TriggerEntry;
42use serde::{Deserialize, Serialize};
43use sha2::{Digest, Sha256};
44use std::io::Read;
45use std::path::Path;
46
47pub const CATALOG_FILENAME: &str = "CATALOG";
48const MAGIC: &[u8; 8] = b"MONGRCAT";
49const CATALOG_FORMAT_VERSION: u16 = 1;
50/// 32-byte meta DEK length (matches [`crate::encryption::DEK_LEN`]).
51pub const META_DEK_LEN: usize = 32;
52
53/// Evaluate a durable-boundary fault-injection hook (spec §9.6, FND-006).
54/// `MongrelError::Other` is the closest existing variant for an injected
55/// fault; a dedicated category arrives with the FND-007 error taxonomy.
56/// Shared by the catalog, snapshot-install, and index-publish hook sites.
57pub(crate) fn inject_hook(name: &'static str) -> Result<()> {
58    mongreldb_fault::inject(name).map_err(|fault| MongrelError::Other(fault.to_string()))
59}
60
61/// Lifecycle state of a catalog table entry.
62#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
63pub enum TableState {
64    /// Live and queryable.
65    Live,
66    /// Logically dropped at `at_epoch`; the physical subdir is reaped once no
67    /// reader pins a snapshot older than `at_epoch`.
68    Dropped { at_epoch: u64 },
69    /// Hidden CTAS build state. It is mounted only by the creating handle and
70    /// becomes queryable through one durable publish operation.
71    Building {
72        intended_name: String,
73        query_id: String,
74        created_at_unix_nanos: u64,
75        #[serde(default)]
76        replaces_table_id: Option<u64>,
77    },
78}
79
80/// One row of the catalog.
81#[derive(Debug, Clone, Serialize, Deserialize)]
82#[serde(deny_unknown_fields)]
83pub struct CatalogEntry {
84    pub table_id: u64,
85    pub name: String,
86    pub schema: Schema,
87    pub state: TableState,
88    pub created_epoch: u64,
89}
90
91/// Persistent definition for a physical materialized-view table.
92#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
93#[serde(deny_unknown_fields)]
94pub struct MaterializedViewEntry {
95    pub name: String,
96    pub query: String,
97    pub last_refresh_epoch: u64,
98    #[serde(default)]
99    pub incremental: Option<IncrementalAggregateView>,
100}
101
102#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
103#[serde(deny_unknown_fields)]
104pub struct IncrementalAggregateView {
105    pub source_table: String,
106    pub source_table_id: u64,
107    pub group_column: u16,
108    pub group_output_column: u16,
109    pub outputs: Vec<IncrementalAggregateOutput>,
110    pub count_output_column: u16,
111    pub checkpoint_event_id: String,
112}
113
114#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
115#[serde(deny_unknown_fields)]
116pub struct IncrementalAggregateOutput {
117    pub output_column: u16,
118    pub kind: IncrementalAggregateKind,
119}
120
121#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
122pub enum IncrementalAggregateKind {
123    Count,
124    Sum { source_column: u16 },
125}
126
127/// The full in-memory catalog, mirrored on disk by [`write_atomic`].
128///
129/// Note: `open_generation` is intentionally **not** stored here — it bumps on
130/// every open, and keeping it in CATALOG would dirty the working tree even for
131/// a bare read. It lives in a separate sidecar file (`_meta/generation`) that
132/// callers can `.gitignore` for content-addressed storage workflows.
133#[derive(Debug, Clone, Serialize, Default)]
134pub struct Catalog {
135    /// Highest epoch ever assigned by this DB's commit sequencer.
136    pub db_epoch: u64,
137    /// Next table id to allocate.
138    pub next_table_id: u64,
139    /// Next shared-WAL segment number to allocate.
140    pub next_segment_no: u64,
141    pub tables: Vec<CatalogEntry>,
142    #[serde(default)]
143    pub procedures: Vec<ProcedureEntry>,
144    #[serde(default)]
145    pub triggers: Vec<TriggerEntry>,
146    #[serde(default)]
147    pub external_tables: Vec<ExternalTableEntry>,
148    #[serde(default)]
149    pub materialized_views: Vec<MaterializedViewEntry>,
150    #[serde(default)]
151    pub security: crate::security::SecurityCatalog,
152    /// Monotonic version for optimistic authorization snapshots.
153    #[serde(default)]
154    pub security_version: u64,
155    /// Catalog-level user accounts (Argon2id-hashed credentials).
156    #[serde(default)]
157    pub users: Vec<crate::auth::UserEntry>,
158    /// Catalog-level role definitions.
159    #[serde(default)]
160    pub roles: Vec<crate::auth::RoleEntry>,
161    /// Next monotonic user id to allocate.
162    #[serde(default)]
163    pub next_user_id: u64,
164    /// When true, every Database/Table/Transaction/MongrelSession operation
165    /// requires an authenticated `Principal` with sufficient permission.
166    /// Defaults to false → existing credentialless databases open unchanged.
167    /// See `docs/15-credential-enforcement.md`.
168    #[serde(default)]
169    pub require_auth: bool,
170    /// SQLite-compatible application metadata used by the SQL layer.
171    #[serde(default, skip_serializing_if = "Option::is_none")]
172    pub user_version: Option<i64>,
173    #[serde(default, skip_serializing_if = "Option::is_none")]
174    pub application_id: Option<i64>,
175    /// Monotonic version of the catalog command state machine (S1F-001).
176    /// `0` = no commands applied (legacy catalog); each applied
177    /// [`crate::catalog_cmds::CatalogCommandRecord`] bumps it by one.
178    #[serde(default, skip_serializing_if = "is_zero_u64")]
179    pub catalog_version: u64,
180    /// Bounded retained tail of applied command records, oldest first.
181    /// Rides inside this checkpoint and every `DdlOp::CatalogSnapshot` WAL
182    /// payload with no on-disk format change. Compacted from the front past
183    /// [`crate::catalog_cmds::COMMAND_HISTORY_LIMIT`].
184    #[serde(default, skip_serializing_if = "Vec::is_empty")]
185    pub command_log: Vec<crate::catalog_cmds::CatalogCommandRecord>,
186    /// Resource-group definitions (S1E-001 forward reference; see
187    /// [`crate::catalog_cmds::ResourceGroupDef`]).
188    #[serde(default, skip_serializing_if = "Vec::is_empty")]
189    pub resource_groups: Vec<crate::catalog_cmds::ResourceGroupDef>,
190    /// Persistent online-job definitions (S1F-002 forward reference; see
191    /// [`crate::catalog_cmds::JobDefinition`]).
192    #[serde(default, skip_serializing_if = "Vec::is_empty")]
193    pub job_definitions: Vec<crate::catalog_cmds::JobDefinition>,
194}
195
196fn is_zero_u64(value: &u64) -> bool {
197    *value == 0
198}
199
200#[derive(Deserialize)]
201#[serde(deny_unknown_fields)]
202struct CatalogWire {
203    db_epoch: u64,
204    next_table_id: u64,
205    next_segment_no: u64,
206    tables: Vec<CatalogEntry>,
207    #[serde(default)]
208    procedures: Vec<ProcedureEntry>,
209    #[serde(default)]
210    triggers: Vec<TriggerEntry>,
211    #[serde(default)]
212    external_tables: Vec<ExternalTableEntry>,
213    #[serde(default)]
214    materialized_views: Vec<MaterializedViewEntry>,
215    #[serde(default)]
216    security: crate::security::SecurityCatalog,
217    #[serde(default)]
218    security_version: u64,
219    #[serde(default)]
220    users: Vec<crate::auth::UserEntry>,
221    #[serde(default)]
222    roles: Vec<crate::auth::RoleEntry>,
223    #[serde(default)]
224    next_user_id: u64,
225    #[serde(default)]
226    require_auth: bool,
227    #[serde(default)]
228    user_version: Option<i64>,
229    #[serde(default)]
230    application_id: Option<i64>,
231    #[serde(default)]
232    catalog_version: u64,
233    #[serde(default)]
234    command_log: Vec<crate::catalog_cmds::CatalogCommandRecord>,
235    #[serde(default)]
236    resource_groups: Vec<crate::catalog_cmds::ResourceGroupDef>,
237    #[serde(default)]
238    job_definitions: Vec<crate::catalog_cmds::JobDefinition>,
239    // Known pre-sidecar field. It is intentionally ignored during migration;
240    // every other unknown top-level field remains rejected.
241    #[serde(default)]
242    open_generation: Option<u64>,
243}
244
245impl<'de> Deserialize<'de> for Catalog {
246    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
247    where
248        D: serde::Deserializer<'de>,
249    {
250        let wire = CatalogWire::deserialize(deserializer)?;
251        let _ = wire.open_generation;
252        Ok(Self {
253            db_epoch: wire.db_epoch,
254            next_table_id: wire.next_table_id,
255            next_segment_no: wire.next_segment_no,
256            tables: wire.tables,
257            procedures: wire.procedures,
258            triggers: wire.triggers,
259            external_tables: wire.external_tables,
260            materialized_views: wire.materialized_views,
261            security: wire.security,
262            security_version: wire.security_version,
263            users: wire.users,
264            roles: wire.roles,
265            next_user_id: wire.next_user_id,
266            require_auth: wire.require_auth,
267            user_version: wire.user_version,
268            application_id: wire.application_id,
269            catalog_version: wire.catalog_version,
270            command_log: wire.command_log,
271            resource_groups: wire.resource_groups,
272            job_definitions: wire.job_definitions,
273        })
274    }
275}
276
277#[derive(Serialize, Deserialize)]
278#[serde(deny_unknown_fields)]
279struct CatalogEnvelope {
280    format_version: u16,
281    catalog: Catalog,
282}
283
284/// The `open_generation` counter, stored in `_meta/generation` (NOT in CATALOG).
285/// Bumped on every open to scope `txn_id` across reopens so ids never alias.
286///
287/// Kept as a sidecar so that CATALOG is stable across bare opens — the only
288/// volatile bytes in the database directory during a read-only session are
289/// this 8-byte file + `.lock` + caches, all of which can be `.gitignore`-d.
290pub const GENERATION_FILENAME: &str = "generation";
291
292/// Read `open_generation` from `_meta/generation`. Missing is reported
293/// separately for one-time migration; malformed or unreadable state fails
294/// closed because resetting this counter can alias retained WAL transactions.
295pub fn read_generation(root: &crate::durable_file::DurableRoot) -> Result<Option<u64>> {
296    let relative = Path::new("_meta").join(GENERATION_FILENAME);
297    match root.entry_exists(&relative) {
298        Ok(true) => {}
299        Ok(false) => return Ok(None),
300        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
301        Err(error) => return Err(error.into()),
302    }
303    let mut file = root.open_regular(&relative)?;
304    let length = file.metadata()?.len();
305    if length != 8 {
306        return Err(MongrelError::Other(format!(
307            "invalid open-generation length: got {length}, expected 8"
308        )));
309    }
310    let mut bytes = [0_u8; 8];
311    std::io::Read::read_exact(&mut file, &mut bytes)?;
312    Ok(Some(u64::from_le_bytes(bytes)))
313}
314
315/// Write `open_generation` to `_meta/generation` atomically (temp + rename +
316/// fsync). This is intentionally a separate file from CATALOG so that CATALOG
317/// stays byte-stable across bare opens (the generation counter is the only
318/// field that changes on every open).
319pub fn write_generation(root: &crate::durable_file::DurableRoot, generation: u64) -> Result<()> {
320    root.create_directory_all("_meta")?;
321    root.write_atomic(
322        Path::new("_meta").join(GENERATION_FILENAME),
323        &generation.to_le_bytes(),
324    )?;
325    Ok(())
326}
327
328impl Catalog {
329    /// An empty catalog for a freshly created DB.
330    pub fn empty() -> Self {
331        Catalog::default()
332    }
333
334    /// Look up an entry by name (live only).
335    pub fn live(&self, name: &str) -> Option<&CatalogEntry> {
336        self.tables
337            .iter()
338            .find(|t| t.name == name && matches!(t.state, TableState::Live))
339    }
340
341    pub(crate) fn building(&self, name: &str) -> Option<&CatalogEntry> {
342        self.tables
343            .iter()
344            .find(|table| table.name == name && matches!(table.state, TableState::Building { .. }))
345    }
346
347    pub(crate) fn building_for(&self, intended_name: &str) -> Option<&CatalogEntry> {
348        self.tables.iter().find(|table| {
349            matches!(
350                &table.state,
351                TableState::Building {
352                    intended_name: candidate,
353                    ..
354                } if candidate == intended_name
355            )
356        })
357    }
358
359    /// Current catalog command state-machine version (S1F-001). `0` means no
360    /// versioned commands have been applied (legacy catalog).
361    pub fn catalog_version(&self) -> u64 {
362        self.catalog_version
363    }
364
365    /// Retained command records with `catalog_version > version`, oldest
366    /// first. History is bounded by
367    /// [`crate::catalog_cmds::COMMAND_HISTORY_LIMIT`]: when the requested
368    /// version predates the retained tail, the result is a strict suffix and
369    /// the caller detects compaction via the first record's `catalog_version`.
370    pub fn commands_since(&self, version: u64) -> Vec<crate::catalog_cmds::CatalogCommandRecord> {
371        self.command_log
372            .iter()
373            .filter(|record| record.catalog_version > version)
374            .cloned()
375            .collect()
376    }
377
378    /// Validate and apply one versioned command record (spec §10.6, S1F-001).
379    ///
380    /// Ordering is fail-closed:
381    ///
382    /// - an unsupported encoding `version` is rejected (§4.10);
383    /// - a record at the current version + 1 is validated, applied
384    ///   deterministically, and appended to the bounded `command_log`;
385    /// - re-applying the identical record at an already-applied version is an
386    ///   idempotent no-op ([`CatalogDelta::NoOp`]); a *different* command
387    ///   claiming an already-applied retained version is a conflict;
388    /// - a record older than the retained window is treated as already
389    ///   applied (compaction makes byte-comparison impossible);
390    /// - any other version (a gap) is a conflict.
391    ///
392    /// This mutates only in-memory state; durability rides the existing
393    /// checkpoint/snapshot mechanics (see the module docs), or use
394    /// [`Catalog::apply_command_and_checkpoint`].
395    pub fn apply_command(
396        &mut self,
397        record: &crate::catalog_cmds::CatalogCommandRecord,
398    ) -> Result<crate::catalog_cmds::CatalogDelta> {
399        use crate::catalog_cmds::{
400            apply, encode_command, CatalogDelta, CATALOG_COMMAND_FORMAT_VERSION,
401            COMMAND_HISTORY_LIMIT,
402        };
403
404        if record.version != CATALOG_COMMAND_FORMAT_VERSION {
405            return Err(MongrelError::UnsupportedStorageVersion {
406                component: "catalog command",
407                found: record.version,
408                supported: CATALOG_COMMAND_FORMAT_VERSION,
409            });
410        }
411        if record.catalog_version <= self.catalog_version {
412            return match self
413                .command_log
414                .iter()
415                .find(|existing| existing.catalog_version == record.catalog_version)
416            {
417                Some(existing) if encode_command(existing)? == encode_command(record)? => {
418                    Ok(CatalogDelta::NoOp)
419                }
420                Some(_) => Err(MongrelError::Conflict(format!(
421                    "a different catalog command is already recorded at version {}",
422                    record.catalog_version
423                ))),
424                // Compacted out of the retained window: treated as applied.
425                None => Ok(CatalogDelta::NoOp),
426            };
427        }
428        let expected = self
429            .catalog_version
430            .checked_add(1)
431            .ok_or_else(|| MongrelError::Full("catalog version space exhausted".into()))?;
432        if record.catalog_version != expected {
433            return Err(MongrelError::Conflict(format!(
434                "catalog command version gap: got {}, expected {expected}",
435                record.catalog_version
436            )));
437        }
438        let delta = apply(self, &record.command)?;
439        delta.apply_to(self)?;
440        self.catalog_version = record.catalog_version;
441        self.command_log.push(record.clone());
442        if self.command_log.len() > COMMAND_HISTORY_LIMIT {
443            let overflow = self.command_log.len() - COMMAND_HISTORY_LIMIT;
444            self.command_log.drain(..overflow);
445        }
446        Ok(delta)
447    }
448
449    /// Apply a versioned command record, then checkpoint the catalog to
450    /// `<dir>/CATALOG` through the existing atomic write path (magic +
451    /// checksum/encryption, rename + directory fsync, `catalog.publish.*`
452    /// fault hooks preserved).
453    pub fn apply_command_and_checkpoint(
454        &mut self,
455        dir: &Path,
456        meta_dek: Option<&[u8; META_DEK_LEN]>,
457        record: &crate::catalog_cmds::CatalogCommandRecord,
458    ) -> Result<crate::catalog_cmds::CatalogDelta> {
459        let delta = self.apply_command(record)?;
460        write_atomic(dir, self, meta_dek)?;
461        Ok(delta)
462    }
463}
464
465fn seal(body: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Vec<u8>> {
466    match meta_dek {
467        Some(dek) => crate::encryption::encrypt_blob(dek, body),
468        None => Ok(plaintext_frame(body)),
469    }
470}
471
472fn plaintext_frame(body: &[u8]) -> Vec<u8> {
473    let hash = Sha256::digest(body);
474    let mut out = Vec::with_capacity(body.len() + 8 + 32);
475    out.extend_from_slice(MAGIC);
476    out.extend_from_slice(&hash);
477    out.extend_from_slice(body);
478    out
479}
480
481/// Atomically write the catalog to `<dir>/CATALOG` (review fix #19: dir-fsync).
482///
483/// If `meta_dek` is `Some`, the body is AES-256-GCM sealed (confidential +
484/// authenticated); otherwise the body carries a SHA-256 tag (integrity only).
485pub fn write_atomic(
486    dir: &Path,
487    cat: &Catalog,
488    meta_dek: Option<&[u8; META_DEK_LEN]>,
489) -> Result<()> {
490    write_atomic_controlled(dir, cat, meta_dek, || Ok(()))
491}
492
493/// Prepare and fsync a complete catalog replacement, then invoke
494/// `before_publish` immediately before the atomic rename makes it durable.
495pub fn write_atomic_controlled<F>(
496    dir: &Path,
497    cat: &Catalog,
498    meta_dek: Option<&[u8; META_DEK_LEN]>,
499    before_publish: F,
500) -> Result<()>
501where
502    F: FnOnce() -> Result<()>,
503{
504    write_atomic_controlled_with_after(dir, cat, meta_dek, before_publish, || {})
505}
506
507/// Controlled catalog replacement with a live-publication callback. The
508/// callback runs after rename, before the parent directory is fsynced. Thus a
509/// later error means the replacement is visible in this process but its
510/// crash-durable outcome is unknown.
511pub(crate) fn write_atomic_controlled_with_after<F, A>(
512    dir: &Path,
513    cat: &Catalog,
514    meta_dek: Option<&[u8; META_DEK_LEN]>,
515    before_publish: F,
516    after_publish: A,
517) -> Result<()>
518where
519    F: FnOnce() -> Result<()>,
520    A: FnOnce(),
521{
522    let body = encode(cat)?;
523    let payload = seal(&body, meta_dek)?;
524
525    // FND-006: fire before the replacement becomes durable.
526    inject_hook("catalog.publish.before")?;
527    let root = crate::durable_file::DurableRoot::open(dir)?;
528    root.write_atomic_controlled_with_after(
529        CATALOG_FILENAME,
530        &payload,
531        before_publish,
532        after_publish,
533    )?;
534    // FND-006: the replacement is durable; the caller still sees a hook
535    // failure as an unknown-outcome error.
536    inject_hook("catalog.publish.after")?;
537    Ok(())
538}
539
540fn open_payload(bytes: &[u8], meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
541    match meta_dek {
542        Some(dek) => match crate::encryption::decrypt_blob(dek, bytes) {
543            Ok(body) => deserialize(&body),
544            Err(_) => Ok(None),
545        },
546        None => parse_plaintext(bytes),
547    }
548}
549
550pub(crate) fn encode(catalog: &Catalog) -> Result<Vec<u8>> {
551    serde_json::to_vec(&CatalogEnvelope {
552        format_version: CATALOG_FORMAT_VERSION,
553        catalog: catalog.clone(),
554    })
555    .map_err(|error| MongrelError::Other(format!("catalog serialize: {error}")))
556}
557
558pub(crate) fn write_durable(
559    root: &crate::durable_file::DurableRoot,
560    catalog: &Catalog,
561    meta_dek: Option<&[u8; META_DEK_LEN]>,
562) -> Result<()> {
563    let body = encode(catalog)?;
564    let payload = seal(&body, meta_dek)?;
565    inject_hook("catalog.publish.before")?;
566    root.write_atomic(CATALOG_FILENAME, &payload)?;
567    inject_hook("catalog.publish.after")?;
568    Ok(())
569}
570
571pub(crate) fn decode(body: &[u8]) -> Result<Catalog> {
572    let value: serde_json::Value = serde_json::from_slice(body)
573        .map_err(|error| MongrelError::Other(format!("catalog deserialize: {error}")))?;
574    let is_envelope = value.as_object().is_some_and(|object| {
575        object.contains_key("format_version") || object.contains_key("catalog")
576    });
577    if !is_envelope {
578        // Legacy pre-envelope catalogs remain readable, but unknown fields are
579        // rejected by `Catalog::deny_unknown_fields`.
580        return serde_json::from_value(value)
581            .map_err(|error| MongrelError::Other(format!("catalog deserialize: {error}")));
582    }
583    let envelope: CatalogEnvelope = serde_json::from_value(value)
584        .map_err(|error| MongrelError::Other(format!("catalog deserialize: {error}")))?;
585    if envelope.format_version != CATALOG_FORMAT_VERSION {
586        return Err(MongrelError::Other(format!(
587            "unsupported catalog format version {}",
588            envelope.format_version
589        )));
590    }
591    Ok(envelope.catalog)
592}
593
594fn deserialize(body: &[u8]) -> Result<Option<Catalog>> {
595    decode(body).map(Some)
596}
597
598/// Read the catalog from `<dir>/CATALOG`. Returns `Ok(None)` if no catalog is
599/// present, or if authentication fails (tampered / wrong key).
600pub fn read(dir: &Path, meta_dek: Option<&[u8; META_DEK_LEN]>) -> Result<Option<Catalog>> {
601    let p = dir.join(CATALOG_FILENAME);
602    let file = match crate::durable_file::open_regular_nofollow(&p) {
603        Ok(file) => file,
604        Err(MongrelError::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
605        Err(e) => return Err(e),
606    };
607    read_file(file, meta_dek)
608}
609
610pub(crate) fn read_durable(
611    root: &crate::durable_file::DurableRoot,
612    meta_dek: Option<&[u8; META_DEK_LEN]>,
613) -> Result<Option<Catalog>> {
614    let file = match root.open_regular(CATALOG_FILENAME) {
615        Ok(file) => file,
616        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
617        Err(error) => return Err(error.into()),
618    };
619    read_file(file, meta_dek)
620}
621
622fn read_file(
623    file: std::fs::File,
624    meta_dek: Option<&[u8; META_DEK_LEN]>,
625) -> Result<Option<Catalog>> {
626    const MAX_CATALOG_BYTES: u64 = 64 * 1024 * 1024;
627    let length = file.metadata()?.len();
628    if length > MAX_CATALOG_BYTES {
629        return Err(MongrelError::ResourceLimitExceeded {
630            resource: "catalog bytes",
631            requested: usize::try_from(length).unwrap_or(usize::MAX),
632            limit: MAX_CATALOG_BYTES as usize,
633        });
634    }
635    let mut bytes = Vec::with_capacity(length as usize);
636    file.take(MAX_CATALOG_BYTES + 1).read_to_end(&mut bytes)?;
637    if bytes.len() as u64 != length {
638        return Err(MongrelError::Other(
639            "catalog length changed while reading".into(),
640        ));
641    }
642    open_payload(&bytes, meta_dek)
643}
644
645fn parse_plaintext(bytes: &[u8]) -> Result<Option<Catalog>> {
646    if bytes.len() < 8 + 32 || &bytes[..8] != MAGIC {
647        return Ok(None);
648    }
649    let (tag, body) = bytes[8..].split_at(32);
650    let calc = Sha256::digest(body);
651    if tag != calc.as_slice() {
652        // tampered
653        return Ok(None);
654    }
655    deserialize(body)
656}
657
658#[cfg(test)]
659mod tests {
660    use super::*;
661
662    #[test]
663    fn empty_catalog_default() {
664        let c = Catalog::empty();
665        assert_eq!(c.db_epoch, 0);
666        assert_eq!(c.next_table_id, 0);
667        assert!(c.tables.is_empty());
668    }
669}
mongreldb_core/catalog.rs

mongreldb_core/
catalog.rs
