feat(storage): read-only direct CompactStore container mmap (G-EM0.1)

crates/grafeo-common/src/utils/error.rs

@@ -66,6 +66,8 @@ pub enum ErrorCode {
     StorageCorrupted,
     /// Recovery from WAL failed.
     StorageRecoveryFailed,
+    /// Requested direct memory mapping is unavailable for this storage layout.
+    StorageDirectMmapUnavailable,
 
     // Validation errors (V)
     /// Request validation failed.
@@ -112,6 +114,7 @@ impl ErrorCode {
             Self::StorageFull => "GRAFEO-S001",
             Self::StorageCorrupted => "GRAFEO-S002",
             Self::StorageRecoveryFailed => "GRAFEO-S003",
+            Self::StorageDirectMmapUnavailable => "GRAFEO-S004",
 
             Self::InvalidInput => "GRAFEO-V001",
             Self::NodeNotFound => "GRAFEO-V002",
@@ -350,6 +353,9 @@ pub enum StorageError {
 
     /// Checkpoint failed.
     CheckpointFailed(String),
+
+    /// The section cannot be served through the direct mapped-read path.
+    DirectMmapUnavailable(String),
 }
 
 impl StorageError {
@@ -361,6 +367,7 @@ impl StorageError {
             Self::Full => ErrorCode::StorageFull,
             Self::InvalidWalEntry(_) | Self::CheckpointFailed(_) => ErrorCode::StorageCorrupted,
             Self::RecoveryFailed(_) => ErrorCode::StorageRecoveryFailed,
+            Self::DirectMmapUnavailable(_) => ErrorCode::StorageDirectMmapUnavailable,
         }
     }
 }
@@ -373,6 +380,9 @@ impl fmt::Display for StorageError {
             StorageError::InvalidWalEntry(msg) => write!(f, "Invalid WAL entry: {msg}"),
             StorageError::RecoveryFailed(msg) => write!(f, "Recovery failed: {msg}"),
             StorageError::CheckpointFailed(msg) => write!(f, "Checkpoint failed: {msg}"),
+            StorageError::DirectMmapUnavailable(msg) => {
+                write!(f, "Direct mmap unavailable: {msg}")
+            }
         }
     }
 }

crates/grafeo-core/src/graph/compact/mod.rs

@@ -38,6 +38,7 @@ pub use builder::{CompactStoreBuilder, from_graph_store, from_graph_store_preser
 use std::sync::Arc;
 
 use arcstr::ArcStr;
+use bytes::Bytes;
 use grafeo_common::types::{EdgeId, NodeId};
 use grafeo_common::utils::hash::FxHashMap;
 
@@ -82,6 +83,14 @@ pub struct CompactStore {
     node_offset_to_id: Option<Vec<Vec<NodeId>>>,
     /// Reverse: rel_table_id index -> vec of original `EdgeId` per CSR position.
     edge_offset_to_id: Option<Vec<Vec<EdgeId>>>,
+    /// Full container mapping retained for a direct mapped read-only reopen.
+    ///
+    /// This is an owner handle only: it does not copy the mapped payload and
+    /// is intentionally excluded from [`Self::memory_bytes`]. The mapped
+    /// graph-view work in G-EM0.2 replaces the remaining proportional decoded
+    /// structures; retaining this handle guarantees no codec-free snapshot can
+    /// accidentally unmap while readers still hold the CompactStore.
+    mapped_backing: Option<Bytes>,
 }
 
 impl std::fmt::Debug for CompactStore {
@@ -149,9 +158,26 @@ impl CompactStore {
             edge_id_map: None,
             node_offset_to_id: None,
             edge_offset_to_id: None,
+            mapped_backing: None,
         }
     }
 
+    /// Retains the owner for a verified direct container mapping.
+    ///
+    /// The mapping is released when this `CompactStore` and every clone of the
+    /// owner `Bytes` have been dropped. Callers must only pass bytes produced
+    /// from a successfully validated immutable container section.
+    pub fn retain_mapped_backing(&mut self, mapped_bytes: Bytes) {
+        self.mapped_backing = Some(mapped_bytes);
+    }
+
+    /// Returns the direct-container mapping length when this store was opened
+    /// from one, excluding it from anonymous heap accounting.
+    #[must_use]
+    pub fn mapped_backing_bytes(&self) -> Option<usize> {
+        self.mapped_backing.as_ref().map(Bytes::len)
+    }
+
     /// Resolves a table_id to its [`NodeTable`].
     #[inline]
     fn resolve_node_table(&self, table_id: u16) -> Option<&NodeTable> {

crates/grafeo-core/src/graph/compact/section.rs

@@ -104,6 +104,33 @@ impl CompactStoreSection {
         Ok(())
     }
 
+    /// Deserializes from a verified direct container mapping.
+    ///
+    /// Unlike [`Self::deserialize_from_bytes`], this retains one owner handle
+    /// on the resulting [`CompactStore`] so a mapping survives even when a
+    /// particular snapshot happens not to contain a column codec slice.
+    /// `data` must originate from the checked `GrafeoFileManager::mmap_section`
+    /// path; callers cannot use this to bypass container CRC validation.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when the mapped payload is truncated, CRC-invalid at
+    /// the CompactStore layer, or otherwise fails the standard v1–v4 codec
+    /// reader. Failures do not expose unchecked slices to the caller.
+    pub fn deserialize_from_mapped_bytes(
+        &mut self,
+        data: bytes::Bytes,
+    ) -> grafeo_common::utils::error::Result<()> {
+        let mut store = deserialize_compact_store(&data).map_err(|e| {
+            grafeo_common::utils::error::Error::Internal(format!(
+                "CompactStore deserialization failed: {e}"
+            ))
+        })?;
+        store.retain_mapped_backing(data);
+        *self.store.write() = Some(Arc::new(store));
+        Ok(())
+    }
+
     /// Serializes at the requested format version.
     ///
     /// The default [`Section::serialize`] always writes [`FORMAT_VERSION`].

crates/grafeo-engine/src/database/mod.rs

@@ -87,6 +87,41 @@ use crate::query::cache::QueryCache;
 use crate::session::Session;
 use crate::transaction::TransactionManager;
 
+/// Actual backing selected for a reopened CompactStore base.
+///
+/// This is intentionally an observed runtime diagnostic rather than a
+/// configuration echo. In particular, `ContainerMmap` is only reported after
+/// `GrafeoFileManager::mmap_section` has verified the selected section CRC and
+/// the CompactStore has accepted its owner-backed bytes.
+#[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum CompactBacking {
+    /// Read-only base is owned by the immutable CompactStore section mapping.
+    ContainerMmap {
+        /// Stable identifier for this immutable container artifact.
+        artifact_id: String,
+        /// CompactStore payload version reported by the verified header.
+        payload_version: u8,
+        /// Bytes mapped from the container section; file-backed, not heap.
+        mapped_bytes: usize,
+    },
+    /// Compatibility path for writable opens and legacy layouts.
+    LegacyEager {
+        /// Stable identifier for the container artifact read eagerly.
+        artifact_id: String,
+        /// CompactStore payload version reported by the verified header.
+        payload_version: u8,
+        /// Complete payload bytes allocated by the compatibility reader.
+        payload_bytes: usize,
+    },
+}
+
+#[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+struct LoadedCompactBase {
+    store: Arc<grafeo_core::graph::compact::CompactStore>,
+    backing: CompactBacking,
+}
+
 /// Your handle to a Grafeo database.
 ///
 /// Start here. Create one with [`new_in_memory()`](Self::new_in_memory) for
@@ -199,6 +234,9 @@ pub struct GrafeoDB {
     /// `layered_store` via `swap_base()`.
     #[cfg(all(feature = "compact-store", feature = "mmap", feature = "lpg"))]
     compact_tiered: Option<Arc<compact_tiered::CompactStoreTiered>>,
+    /// Observed compact-base backing after a persistent reopen.
+    #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+    compact_backing: Option<CompactBacking>,
 }
 
 impl GrafeoDB {
@@ -393,9 +431,7 @@ impl GrafeoDB {
         // compacted database. The post-construction wiring uses this to
         // rebuild the LayeredStore + tier wrapper + overlay consumer.
         #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
-        let mut loaded_compact_base: Option<
-            Arc<grafeo_core::graph::compact::CompactStore>,
-        > = None;
+        let mut loaded_compact_base: Option<LoadedCompactBase> = None;
 
         // Phase 5e: snapshot of the OverlayDeletions section (if present),
         // applied after the LayeredStore is wired so that previously-deleted
@@ -425,7 +461,7 @@ impl GrafeoDB {
                         )?;
                         #[cfg(feature = "compact-store")]
                         {
-                            loaded_compact_base = Self::extract_compact_base(&fm)?;
+                            loaded_compact_base = Self::extract_compact_base(&fm, true)?;
                             loaded_overlay_deletions = Self::extract_overlay_deletions(&fm)?;
                         }
                     } else {
@@ -483,7 +519,7 @@ impl GrafeoDB {
                     )?;
                     #[cfg(feature = "compact-store")]
                     {
-                        loaded_compact_base = Self::extract_compact_base(&fm)?;
+                        loaded_compact_base = Self::extract_compact_base(&fm, false)?;
                         loaded_overlay_deletions = Self::extract_overlay_deletions(&fm)?;
                     }
                 } else {
@@ -661,6 +697,8 @@ impl GrafeoDB {
             layered_store: None,
             #[cfg(all(feature = "compact-store", feature = "mmap", feature = "lpg"))]
             compact_tiered: None,
+            #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+            compact_backing: None,
         };
 
         // Register storage sections as memory consumers for pressure tracking
@@ -672,8 +710,9 @@ impl GrafeoDB {
         // engine sees the full picture and the read/write paths route
         // through the layered store.
         #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
-        if let Some(compact_base) = loaded_compact_base {
-            db.wire_layered_after_load(compact_base, loaded_overlay_deletions)?;
+        if let Some(loaded) = loaded_compact_base {
+            db.compact_backing = Some(loaded.backing);
+            db.wire_layered_after_load(loaded.store, loaded_overlay_deletions)?;
         }
 
         // After Catalog shells + VectorStore topology + WAL + layered wiring,
@@ -815,6 +854,8 @@ impl GrafeoDB {
             layered_store: None,
             #[cfg(all(feature = "compact-store", feature = "mmap", feature = "lpg"))]
             compact_tiered: None,
+            #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+            compact_backing: None,
         })
     }
 
@@ -906,6 +947,8 @@ impl GrafeoDB {
             layered_store: None,
             #[cfg(all(feature = "compact-store", feature = "mmap", feature = "lpg"))]
             compact_tiered: None,
+            #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+            compact_backing: None,
         })
     }
 
@@ -1524,21 +1567,59 @@ impl GrafeoDB {
     /// section file, if present. Used by the open path to reconstruct
     /// the LayeredStore wiring after a previously-compacted database
     /// reopens.
+    ///
+    /// When `direct_mmap` is true (read-only open), this path never calls
+    /// [`GrafeoFileManager::read_section_data`] and never copies the full
+    /// section into an owned payload buffer. Mapping failures — including
+    /// encrypted layouts — fail closed with
+    /// [`StorageError::DirectMmapUnavailable`](grafeo_common::utils::error::StorageError::DirectMmapUnavailable)
+    /// rather than silently falling back to eager materialization.
+    /// Writable opens keep the legacy eager path and report
+    /// [`CompactBacking::LegacyEager`]; that is not Milestone R evidence.
     #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
     fn extract_compact_base(
         fm: &GrafeoFileManager,
-    ) -> Result<Option<Arc<grafeo_core::graph::compact::CompactStore>>> {
+        direct_mmap: bool,
+    ) -> Result<Option<LoadedCompactBase>> {
         use grafeo_common::storage::{Section, SectionType};
         let Some(dir) = fm.read_section_directory()? else {
             return Ok(None);
         };
         let Some(entry) = dir.find(SectionType::CompactStore) else {
             return Ok(None);
         };
-        let data = fm.read_section_data(entry)?;
+        let artifact_id = format!("container:{}:{:08x}", fm.path().display(), entry.checksum);
         let mut section = grafeo_core::graph::compact::section::CompactStoreSection::empty();
-        section.deserialize(&data)?;
-        Ok(section.store())
+        let backing = if direct_mmap {
+            let mapped = Arc::new(fm.mmap_section(entry)?);
+            let mapped_bytes = mapped.len();
+            let data = mapped.into_bytes();
+            let payload_version = data.get(4).copied().ok_or_else(|| {
+                Error::Internal("truncated CompactStore payload after mmap validation".into())
+            })?;
+            section.deserialize_from_mapped_bytes(data)?;
+            CompactBacking::ContainerMmap {
+                artifact_id,
+                payload_version,
+                mapped_bytes,
+            }
+        } else {
+            let data = fm.read_section_data(entry)?;
+            let payload_version = data.get(4).copied().ok_or_else(|| {
+                Error::Internal("truncated CompactStore payload after read validation".into())
+            })?;
+            let payload_bytes = data.len();
+            section.deserialize(&data)?;
+            CompactBacking::LegacyEager {
+                artifact_id,
+                payload_version,
+                payload_bytes,
+            }
+        };
+        let store = section.store().ok_or_else(|| {
+            Error::Internal("CompactStore section deserialized without a store".into())
+        })?;
+        Ok(Some(LoadedCompactBase { store, backing }))
     }
 
     /// Reads the persisted overlay deletion log from the container, if
@@ -2365,6 +2446,18 @@ impl GrafeoDB {
         self.compact_tiered.as_ref()
     }
 
+    /// Returns the actual backing selected for a reopened CompactStore base.
+    ///
+    /// `ContainerMmap` proves that this handle owns the immutable container
+    /// mapping directly. It is distinct from the legacy spill-sidecar tier and
+    /// from a directory `mmap_able` flag, neither of which proves the normal
+    /// reopen path avoided an owned full-section read.
+    #[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+    #[must_use]
+    pub fn compact_backing(&self) -> Option<&CompactBacking> {
+        self.compact_backing.as_ref()
+    }
+
     /// Returns the query cache.
     #[must_use]
     pub fn query_cache(&self) -> &Arc<QueryCache> {

crates/grafeo-engine/src/lib.rs

@@ -51,6 +51,8 @@ pub use admin::{
 pub use auth::{Grant, Identity, Role, StatementKind};
 pub use catalog::{Catalog, CatalogError, IndexDefinition, IndexType};
 pub use config::{AccessMode, Config, ConfigError, DurabilityMode, GraphModel};
+#[cfg(all(feature = "grafeo-file", feature = "lpg", feature = "compact-store"))]
+pub use database::CompactBacking;
 pub use database::GrafeoDB;
 #[cfg(all(feature = "lpg", feature = "vector-index"))]
 pub use database::IndexedVectorRead;