pub mod encode;
pub mod layout;
pub mod seam;
use crate::types::{GraphError, Result};
use crate::v8::layout::{
ArchivedColumns, ArchivedCsr, ArchivedEdgeProps, ArchivedHnsw, ArchivedIdMap, ArchivedInterner,
ArchivedRulesMeta, ArchivedViews,
};
use memmap2::MmapOptions;
use std::path::Path;
use std::sync::atomic::{AtomicU8, Ordering};
enum Backing {
Mapped(memmap2::Mmap),
Owned(Vec<u8>),
}
impl std::ops::Deref for Backing {
type Target = [u8];
fn deref(&self) -> &[u8] {
match self {
Backing::Mapped(m) => m.as_ref(),
Backing::Owned(v) => v.as_slice(),
}
}
}
pub const HEADER_SIZE: usize = 4096;
pub const SECTION_TOPOLOGY: u8 = 0;
pub const SECTION_COLUMNS: u8 = 1;
pub const SECTION_IDS: u8 = 2;
pub const SECTION_SYMS: u8 = 3;
pub const SECTION_META: u8 = 4;
pub const SECTION_EDGE_PROPS: u8 = 5;
pub const SECTION_HNSW: u8 = 6;
pub const SECTION_PROVENANCE: u8 = 7;
pub const SECTION_RULES_META: u8 = 8;
pub const SECTION_VIEWS: u8 = 9;
pub const SECTION_IVF_STATE: u8 = 10;
pub const SECTION_LAST_CHANGE: u8 = 11;
pub const V8_MAGIC_SECTION_COUNT: usize = 12;
fn is_large_section(id: u8) -> bool {
matches!(
id,
SECTION_TOPOLOGY
| SECTION_COLUMNS
| SECTION_EDGE_PROPS
| SECTION_HNSW
| SECTION_PROVENANCE
| SECTION_IVF_STATE
)
}
const STATE_UNCHECKED: u8 = 0;
const STATE_OK: u8 = 1;
const STATE_BAD: u8 = 2;
#[derive(Clone, Copy)]
struct SectionEntry {
id: u8,
offset: u32,
len: u32,
crc32: u32,
}
pub struct MappedBase {
backing: Backing,
dir: Vec<SectionEntry>,
check_state: [AtomicU8; V8_MAGIC_SECTION_COUNT],
mixed: crate::v8::seam::MixedCache,
}
impl MappedBase {
pub fn map(path: &Path) -> Result<Self> {
let file = std::fs::File::open(path).map_err(GraphError::Io)?;
let mmap = unsafe { MmapOptions::new().map(&file) }.map_err(GraphError::Io)?;
let dir = parse_header(&mmap)?;
Ok(Self {
backing: Backing::Mapped(mmap),
dir,
check_state: std::array::from_fn(|_| AtomicU8::new(STATE_UNCHECKED)),
mixed: Default::default(),
})
}
pub fn from_bytes(bytes: Vec<u8>) -> Result<Self> {
let dir = parse_header(&bytes)?;
Ok(Self {
backing: Backing::Owned(bytes),
dir,
check_state: std::array::from_fn(|_| AtomicU8::new(STATE_UNCHECKED)),
mixed: Default::default(),
})
}
pub fn validate_section_bounds(&self) -> Result<()> {
for entry in &self.dir {
let start = entry.offset as usize;
let end = start
.checked_add(entry.len as usize)
.ok_or_else(|| GraphError::Corrupt {
detail: format!(
"v8: section {} length overflow (offset={}, len={})",
entry.id, entry.offset, entry.len
),
})?;
self.backing
.get(start..end)
.ok_or_else(|| GraphError::Corrupt {
detail: format!(
"v8: section {} extends beyond file (end={}, file_len={})",
entry.id,
end,
self.backing.len()
),
})?;
if let Some(min) = min_rkyv_root_size(entry.id) {
if (entry.len as usize) < min {
return Err(GraphError::Corrupt {
detail: format!(
"v8: section {} payload too small for rkyv root \
(len={}, minimum={})",
entry.id, entry.len, min
),
});
}
}
}
Ok(())
}
pub fn verify_integrity(
&self,
) -> Vec<(u8, &'static str, usize, std::result::Result<(), String>)> {
let name = |id| match id {
SECTION_TOPOLOGY => "topology",
SECTION_COLUMNS => "columns",
SECTION_IDS => "ids",
SECTION_SYMS => "syms",
SECTION_META => "meta",
SECTION_EDGE_PROPS => "edge_props",
SECTION_HNSW => "hnsw",
SECTION_PROVENANCE => "provenance",
SECTION_RULES_META => "rules_meta",
SECTION_VIEWS => "views",
SECTION_IVF_STATE => "ivf_state",
SECTION_LAST_CHANGE => "last_change",
_ => "unknown",
};
self.dir
.iter()
.map(|entry| {
let id = entry.id;
let start = entry.offset as usize;
let end = match start.checked_add(entry.len as usize) {
Some(e) => e,
None => {
return (
id,
name(id),
0,
Err(format!("section {id}: length overflow")),
)
}
};
let bytes = match self.backing.get(start..end) {
Some(b) => b,
None => {
return (
id,
name(id),
0,
Err(format!("section {id}: extends beyond file")),
)
}
};
let computed = crc32fast::hash(bytes);
if computed != entry.crc32 {
(
id,
name(id),
bytes.len(),
Err(format!(
"CRC mismatch (expected {:08x}, computed {:08x})",
entry.crc32, computed
)),
)
} else {
(id, name(id), bytes.len(), Ok(()))
}
})
.collect()
}
pub(crate) fn section_bytes(&self, section_id: u8) -> Result<&[u8]> {
let entry = self
.dir
.iter()
.find(|e| e.id == section_id)
.ok_or_else(|| GraphError::Corrupt {
detail: format!("v8: section {section_id} not found in directory"),
})?;
let start = entry.offset as usize;
let end = start
.checked_add(entry.len as usize)
.ok_or_else(|| GraphError::Corrupt {
detail: format!("v8: section {section_id} length overflow"),
})?;
let bytes = self
.backing
.get(start..end)
.ok_or_else(|| GraphError::Corrupt {
detail: format!("v8: section {section_id} extends beyond file"),
})?;
let _trace_t = if std::env::var("MUSHROOMDB_TRACE_OPEN").is_ok() {
Some((section_id, std::time::Instant::now()))
} else {
None
};
if !is_large_section(section_id) {
let idx = section_id as usize;
debug_assert!(
idx < V8_MAGIC_SECTION_COUNT,
"section_id {section_id} >= V8_MAGIC_SECTION_COUNT ({V8_MAGIC_SECTION_COUNT}); \
resize check_state before adding new section ids"
);
if idx < V8_MAGIC_SECTION_COUNT {
match self.check_state[idx].load(Ordering::Acquire) {
STATE_OK => {} STATE_BAD => {
return Err(GraphError::Corrupt {
detail: format!("v8: section {section_id} CRC mismatch (cached)"),
});
}
_ => {
let computed = crc32fast::hash(bytes);
if computed != entry.crc32 {
self.check_state[idx].store(STATE_BAD, Ordering::Release);
return Err(GraphError::Corrupt {
detail: format!(
"v8: section {section_id} CRC mismatch \
(expected {:08x}, computed {:08x})",
entry.crc32, computed
),
});
}
self.check_state[idx].store(STATE_OK, Ordering::Release);
}
}
}
}
if let Some((id, t)) = _trace_t {
eprintln!(
"[MUSHROOMDB_TRACE_OPEN] section_bytes({id}): {:>9.3?}",
t.elapsed()
);
}
Ok(bytes)
}
pub fn topology(&self) -> Result<&ArchivedCsr> {
let bytes = self.section_bytes(SECTION_TOPOLOGY)?;
if bytes.len() < std::mem::size_of::<crate::v8::layout::ArchivedCsrData>() {
return Err(GraphError::Corrupt {
detail: "v8: topology section too short for rkyv root".to_string(),
});
}
Ok(unsafe { rkyv::access_unchecked::<crate::v8::layout::ArchivedCsrData>(bytes) })
}
pub fn mixed_cache(&self) -> &crate::v8::seam::MixedCache {
&self.mixed
}
pub fn columns(&self) -> Result<&ArchivedColumns> {
let bytes = self.section_bytes(SECTION_COLUMNS)?;
if bytes.len() < std::mem::size_of::<crate::v8::layout::ArchivedColumnsData>() {
return Err(GraphError::Corrupt {
detail: "v8: columns section too short for rkyv root".to_string(),
});
}
Ok(unsafe { rkyv::access_unchecked::<crate::v8::layout::ArchivedColumnsData>(bytes) })
}
pub fn ids(&self) -> Result<&ArchivedIdMap> {
let bytes = self.section_bytes(SECTION_IDS)?;
rkyv::access::<crate::v8::layout::ArchivedIdMapData, rkyv::rancor::Error>(bytes).map_err(
|e| GraphError::Corrupt {
detail: format!("v8: ids rkyv access: {e}"),
},
)
}
pub fn syms(&self) -> Result<&ArchivedInterner> {
let bytes = self.section_bytes(SECTION_SYMS)?;
rkyv::access::<crate::v8::layout::ArchivedInternerData, rkyv::rancor::Error>(bytes).map_err(
|e| GraphError::Corrupt {
detail: format!("v8: syms rkyv access: {e}"),
},
)
}
pub fn meta_bytes(&self) -> Result<&[u8]> {
self.section_bytes(SECTION_META)
}
pub fn edge_props_section(&self) -> Result<&ArchivedEdgeProps> {
let bytes = self.section_bytes(SECTION_EDGE_PROPS)?;
if bytes.len() < std::mem::size_of::<crate::v8::layout::ArchivedEdgePropsData>() {
return Err(GraphError::Corrupt {
detail: "v8: edge_props section too short for rkyv root".to_string(),
});
}
Ok(unsafe { rkyv::access_unchecked::<crate::v8::layout::ArchivedEdgePropsData>(bytes) })
}
pub fn hnsw_section(&self) -> Result<&ArchivedHnsw> {
let bytes = self.section_bytes(SECTION_HNSW)?;
if bytes.len() < std::mem::size_of::<crate::v8::layout::ArchivedHnswSectionData>() {
return Err(GraphError::Corrupt {
detail: "v8: hnsw section too short for rkyv root".to_string(),
});
}
Ok(unsafe { rkyv::access_unchecked::<crate::v8::layout::ArchivedHnswSectionData>(bytes) })
}
pub fn validate_hot_sections(&self) -> Result<()> {
use crate::v8::layout::{
ArchivedColumnsData, ArchivedCsrData, ArchivedEdgePropsData, ArchivedHnswSectionData,
};
let check = |bytes: &[u8], name: &str| -> Result<()> {
match name {
"topology" => {
rkyv::access::<ArchivedCsrData, rkyv::rancor::Error>(bytes).map(|_| ())
}
"columns" => {
rkyv::access::<ArchivedColumnsData, rkyv::rancor::Error>(bytes).map(|_| ())
}
"edge_props" => {
rkyv::access::<ArchivedEdgePropsData, rkyv::rancor::Error>(bytes).map(|_| ())
}
"hnsw" => {
rkyv::access::<ArchivedHnswSectionData, rkyv::rancor::Error>(bytes).map(|_| ())
}
_ => Ok(()),
}
.map_err(|e| GraphError::Corrupt {
detail: format!("v8: {name} section failed structural validation: {e}"),
})
};
check(self.section_bytes(SECTION_TOPOLOGY)?, "topology")?;
check(self.section_bytes(SECTION_COLUMNS)?, "columns")?;
check(self.section_bytes(SECTION_EDGE_PROPS)?, "edge_props")?;
check(self.section_bytes(SECTION_HNSW)?, "hnsw")?;
Ok(())
}
pub fn rules_meta_section(&self) -> Result<&ArchivedRulesMeta> {
let bytes = self.section_bytes(SECTION_RULES_META)?;
rkyv::access::<crate::v8::layout::ArchivedRulesMetaData, rkyv::rancor::Error>(bytes)
.map_err(|e| GraphError::Corrupt {
detail: format!("v8: rules_meta rkyv access: {e}"),
})
}
pub fn views_section(&self) -> Result<&ArchivedViews> {
let bytes = self.section_bytes(SECTION_VIEWS)?;
rkyv::access::<crate::v8::layout::ArchivedViewsSectionData, rkyv::rancor::Error>(bytes)
.map_err(|e| GraphError::Corrupt {
detail: format!("v8: views rkyv access: {e}"),
})
}
pub fn ivf_bytes(&self) -> Result<&[u8]> {
if self.dir.iter().all(|e| e.id != SECTION_IVF_STATE) {
return Ok(&[]);
}
self.section_bytes(SECTION_IVF_STATE)
}
pub fn last_change_bytes(&self) -> Result<&[u8]> {
if self.dir.iter().all(|e| e.id != SECTION_LAST_CHANGE) {
return Ok(&[]);
}
self.section_bytes(SECTION_LAST_CHANGE)
}
pub fn edge_props_raw_bytes(&self) -> Result<&[u8]> {
self.section_bytes(SECTION_EDGE_PROPS)
}
pub fn provenance_raw_bytes(&self) -> Result<&[u8]> {
self.section_bytes(SECTION_PROVENANCE)
}
}
fn min_rkyv_root_size(section_id: u8) -> Option<usize> {
use crate::v8::layout::{
ArchivedColumnsData, ArchivedCsrData, ArchivedEdgePropsData, ArchivedHnswSectionData,
};
match section_id {
SECTION_TOPOLOGY => Some(std::mem::size_of::<ArchivedCsrData>()),
SECTION_COLUMNS => Some(std::mem::size_of::<ArchivedColumnsData>()),
SECTION_EDGE_PROPS => Some(std::mem::size_of::<ArchivedEdgePropsData>()),
SECTION_HNSW => Some(std::mem::size_of::<ArchivedHnswSectionData>()),
_ => None,
}
}
fn parse_header(mmap: &[u8]) -> Result<Vec<SectionEntry>> {
if mmap.len() < HEADER_SIZE {
return Err(GraphError::Corrupt {
detail: format!(
"v8: file is {} bytes; minimum for header is {HEADER_SIZE}",
mmap.len()
),
});
}
if &mmap[0..4] != b"GDB1" {
return Err(GraphError::Corrupt {
detail: "v8: bad magic (expected GDB1)".into(),
});
}
let version = u16::from_le_bytes(mmap[4..6].try_into().unwrap());
if version != 8 {
return Err(GraphError::Corrupt {
detail: format!("v8: expected version 8, got {version}"),
});
}
let section_count = u16::from_le_bytes(mmap[6..8].try_into().unwrap()) as usize;
let dir_end = 8usize
.checked_add(section_count.saturating_mul(16))
.ok_or_else(|| GraphError::Corrupt {
detail: "v8: directory length overflow".into(),
})?;
if dir_end + 4 > HEADER_SIZE {
return Err(GraphError::Corrupt {
detail: format!(
"v8: {section_count} sections require dir_end={dir_end} which overflows the header"
),
});
}
let stored_crc = u32::from_le_bytes(mmap[dir_end..dir_end + 4].try_into().unwrap());
let computed_crc = crc32fast::hash(&mmap[0..dir_end]);
if stored_crc != computed_crc {
return Err(GraphError::Corrupt {
detail: format!(
"v8: header CRC mismatch (expected {:08x}, computed {:08x})",
stored_crc, computed_crc
),
});
}
let mut dir = Vec::with_capacity(section_count);
for i in 0..section_count {
let base = 8 + i * 16;
let id = mmap[base];
let offset = u32::from_le_bytes(mmap[base + 4..base + 8].try_into().unwrap());
let len = u32::from_le_bytes(mmap[base + 8..base + 12].try_into().unwrap());
let crc32 = u32::from_le_bytes(mmap[base + 12..base + 16].try_into().unwrap());
dir.push(SectionEntry {
id,
offset,
len,
crc32,
});
}
Ok(dir)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::columns::ColumnStore;
use crate::idmap::IdMap;
use crate::interner::Interner;
use crate::topology::Topology;
use crate::types::Value;
use crate::v8::encode::{encode_v8, V8Meta};
use std::collections::{BTreeMap, HashMap};
fn tiny_v8_meta() -> V8Meta {
V8Meta {
labels: vec![0, 0],
edge_props: crate::edge_props::EdgeProps::new(),
rule_defs: vec![],
provenance: BTreeMap::new(),
rule_tripped: BTreeMap::new(),
rule_fires: BTreeMap::new(),
ivf_bytes: Vec::new(),
view_defs: vec![],
wal_truncated: false,
hnsw: BTreeMap::new(),
last_change: HashMap::new(),
}
}
fn encode_tiny() -> Vec<u8> {
let mut ids = IdMap::new();
ids.get_or_insert("a");
ids.get_or_insert("b");
let mut syms = Interner::new();
let e = syms.intern("E");
let mut topo = Topology::new();
topo.add_edge(e, 0, 1);
let mut props = ColumnStore::new();
props.set(0, "v", Value::Int(42));
let meta = tiny_v8_meta();
let mut out = Vec::new();
encode_v8(
None, None, None, None, &topo, &props, &ids, &syms, &meta, &mut out,
)
.expect("encode_v8");
out
}
fn tmp_path(suffix: &str) -> std::path::PathBuf {
std::path::PathBuf::from(format!(
"/tmp/mushroom_v8_{}_{}.bin",
std::process::id(),
suffix
))
}
#[test]
fn v8_encode_and_map_sections_valid() {
let bytes = encode_tiny();
let path = tmp_path("valid");
std::fs::write(&path, &bytes).unwrap();
let _cleanup = defer_remove(&path);
let base = MappedBase::map(&path).expect("map");
let topo = base.topology().expect("topology()");
assert_eq!(u64::from(topo.edge_count), 1);
let ids = base.ids().expect("ids()");
assert_eq!(ids.to_key.len(), 2);
let syms = base.syms().expect("syms()");
assert_eq!(syms.to_str.len(), 1);
assert_eq!(syms.to_str[0].as_str(), "E");
}
#[test]
fn v8_corrupt_section_crc_returns_corrupt_error() {
let mut bytes = encode_tiny();
let section_count = u16::from_le_bytes(bytes[6..8].try_into().unwrap()) as usize;
let mut target_entry_base = None;
for i in 0..section_count {
let base = 8 + i * 16;
if bytes[base] == SECTION_IDS {
target_entry_base = Some(base);
break;
}
}
let entry_base = target_entry_base.expect("SECTION_IDS not found in encode_tiny output");
let section_offset =
u32::from_le_bytes(bytes[entry_base + 4..entry_base + 8].try_into().unwrap()) as usize;
if section_offset < bytes.len() {
bytes[section_offset] ^= 0xff;
}
let path = tmp_path("corrupt");
std::fs::write(&path, &bytes).unwrap();
let _cleanup = defer_remove(&path);
match MappedBase::map(&path) {
Ok(base) => {
let result = base.ids();
match result {
Err(GraphError::Corrupt { .. }) => {}
Err(e) => panic!("expected Corrupt, got {e:?}"),
Ok(_) => panic!("expected Corrupt error but ids() succeeded"),
}
}
Err(GraphError::Corrupt { .. }) => {
}
Err(e) => panic!("unexpected error: {e:?}"),
}
}
#[test]
fn v8_verify_integrity_detects_large_section_corruption() {
let mut bytes = encode_tiny();
let section_count = u16::from_le_bytes(bytes[6..8].try_into().unwrap()) as usize;
let mut target_entry_base = None;
for i in 0..section_count {
let base = 8 + i * 16;
if bytes[base] == SECTION_TOPOLOGY {
target_entry_base = Some(base);
break;
}
}
let entry_base = target_entry_base.expect("SECTION_TOPOLOGY not found");
let section_offset =
u32::from_le_bytes(bytes[entry_base + 4..entry_base + 8].try_into().unwrap()) as usize;
if section_offset < bytes.len() {
bytes[section_offset] ^= 0xff;
}
let path = tmp_path("corrupt_large");
std::fs::write(&path, &bytes).unwrap();
let _cleanup = defer_remove(&path);
let base = MappedBase::map(&path).expect("map");
let _ = base
.topology()
.expect("topology() must not CRC-fail large section");
let results = base.verify_integrity();
let topo = results
.iter()
.find(|(id, _, _, _)| *id == SECTION_TOPOLOGY)
.expect("topology entry in verify results");
assert!(
topo.3.is_err(),
"verify_integrity must detect TOPOLOGY corruption; got Ok"
);
}
#[test]
fn validate_section_bounds_rejects_tiny_section_len() {
let mut bytes = encode_tiny();
let section_count = u16::from_le_bytes(bytes[6..8].try_into().unwrap()) as usize;
let mut topo_entry_base = None;
for i in 0..section_count {
let base = 8 + i * 16;
if bytes[base] == SECTION_TOPOLOGY {
topo_entry_base = Some(base);
break;
}
}
let entry_base = topo_entry_base.expect("SECTION_TOPOLOGY in directory");
let tiny_len: u32 = 1;
bytes[entry_base + 8..entry_base + 12].copy_from_slice(&tiny_len.to_le_bytes());
let dir_end = 8 + section_count * 16;
let new_crc = crc32fast::hash(&bytes[0..dir_end]);
bytes[dir_end..dir_end + 4].copy_from_slice(&new_crc.to_le_bytes());
let base = MappedBase::from_bytes(bytes).expect("header CRC is correct after recompute");
let result = base.validate_section_bounds();
match result {
Err(GraphError::Corrupt { detail }) => {
assert!(
detail.contains("too small for rkyv root") || detail.contains("section"),
"error should mention tiny section; got: {detail}"
);
}
Err(other) => panic!("expected Corrupt, got {other:?}"),
Ok(_) => panic!("expected Err(Corrupt) for tiny section len, got Ok"),
}
}
struct DeferRemove(std::path::PathBuf);
impl Drop for DeferRemove {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.0);
}
}
fn defer_remove(p: &std::path::Path) -> DeferRemove {
DeferRemove(p.to_path_buf())
}
}