From 4845b480f4b15bcb8789d077b99ba68291dd9ffc Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 07:32:53 +0000 Subject: [PATCH 1/6] feat(store): open kernel databases safely Acquire the writer lease before database access, classify foreign files from the header, and quarantine only conclusive kernel mismatches. Exact opens verify the frozen schema before enabling WAL and stamping the writer fence. --- Cargo.lock | 1 + crates/mc-store/Cargo.toml | 1 + crates/mc-store/src/kernel/mod.rs | 3 + crates/mc-store/src/kernel/open.rs | 693 ++++++++++++++++++++++++++ crates/mc-store/src/kernel/schema.rs | 32 +- crates/mc-store/src/sqlite_runtime.rs | 18 +- crates/mc-store/tests/kernel_open.rs | 315 ++++++++++++ 7 files changed, 1058 insertions(+), 5 deletions(-) create mode 100644 crates/mc-store/src/kernel/open.rs create mode 100644 crates/mc-store/tests/kernel_open.rs diff --git a/Cargo.lock b/Cargo.lock index 34e4ec484..41c7dd408 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1045,6 +1045,7 @@ name = "mc-store" version = "0.1.0" dependencies = [ "cortexkit-cache-core", + "cortexkit-lease", "cortexkit-store", "cortexkit-store-types", "flate2", diff --git a/crates/mc-store/Cargo.toml b/crates/mc-store/Cargo.toml index 51ac1e085..f91e5e6d8 100644 --- a/crates/mc-store/Cargo.toml +++ b/crates/mc-store/Cargo.toml @@ -14,6 +14,7 @@ mc-core = { workspace = true } cortexkit-cache-core = { workspace = true } cortexkit-store = { workspace = true } cortexkit-store-types = { workspace = true } +cortexkit-lease = { workspace = true } rusqlite = { workspace = true, features = ["functions"] } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/mc-store/src/kernel/mod.rs b/crates/mc-store/src/kernel/mod.rs index 1ce7e1766..a869c8539 100644 --- a/crates/mc-store/src/kernel/mod.rs +++ b/crates/mc-store/src/kernel/mod.rs @@ -1 +1,4 @@ +mod open; pub mod schema; + +pub use open::{KernelError, KernelStore}; diff --git a/crates/mc-store/src/kernel/open.rs b/crates/mc-store/src/kernel/open.rs new file mode 100644 index 000000000..b50ee99a6 --- /dev/null +++ b/crates/mc-store/src/kernel/open.rs @@ -0,0 +1,693 @@ +use cortexkit_lease::{ + protect_file, FileLeaseStore, LeaseError, LeaseHandle, LeaseKey, LeaseStore, +}; +use rusqlite::{Connection, OpenFlags, OptionalExtension}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::fmt; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Mutex; + +use super::schema::{ + apply_kernel_schema, kernel_schema_digest, kernel_schema_object_inventory, + KERNEL_APPLICATION_ID, KERNEL_FORMAT_EPOCH, +}; +use crate::sqlite_runtime::{ + compute_marker_digest_for_application_id, evaluate_sqlite_runtime_gate, + probe_sqlite_engine_identity_off_path, SqliteEngineIdentity, +}; + +const BUSY_TIMEOUT_MS: i64 = 5_000; +const READ_POOL_SIZE: usize = 2; +const RESET_MARKER_PROTOCOL: &str = "mc-kernel-reset-marker-v1"; +const RESET_MARKER_SUFFIX: &str = ".mc-reset"; +const QUARANTINE_INFIX: &str = ".mc-quarantine-"; +const SQLITE_HEADER: &[u8; 16] = b"SQLite format 3\0"; + +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum KernelError { + Held, + EngineUnsupported, + Foreign, + Inconclusive, + Io, + IdentityMismatch, + FenceLost, +} + +impl fmt::Display for KernelError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Held => "kernel store is held by another writer", + Self::EngineUnsupported => "SQLite engine is unsupported for the kernel store", + Self::Foreign => "kernel store path contains a foreign database family", + Self::Inconclusive => "kernel store identity could not be established safely", + Self::Io => "kernel store I/O failed", + Self::IdentityMismatch => "kernel store identity does not match this build", + Self::FenceLost => "kernel store writer fence was lost", + }) + } +} + +impl fmt::Debug for KernelError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, f) + } +} + +impl std::error::Error for KernelError {} + +pub struct KernelStore { + writer: Mutex, + readers: Vec>, + next_reader: AtomicUsize, + lease_epoch: u64, + _lease: Box, +} + +impl fmt::Debug for KernelStore { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("KernelStore") + .field("lease_epoch", &self.lease_epoch) + .field("read_pool_size", &self.readers.len()) + .finish_non_exhaustive() + } +} + +impl KernelStore { + pub fn open(root: impl AsRef) -> Result { + let identity = + probe_sqlite_engine_identity_off_path().map_err(|_| KernelError::EngineUnsupported)?; + Self::open_with_engine_identity(root, &identity) + } + + fn open_with_engine_identity( + root: impl AsRef, + identity: &SqliteEngineIdentity, + ) -> Result { + if !evaluate_sqlite_runtime_gate(identity).is_empty() { + return Err(KernelError::EngineUnsupported); + } + Self::open_supported(root) + } + + #[cfg(feature = "test-support")] + pub fn open_with_engine_identity_for_test( + root: impl AsRef, + identity: &SqliteEngineIdentity, + ) -> Result { + Self::open_with_engine_identity(root, identity) + } + + fn open_supported(root: impl AsRef) -> Result { + let root = absolute_path(root.as_ref())?; + prepare_root(&root)?; + let db_path = root.join("core.sqlite"); + let lease_store = FileLeaseStore::new(root.join("leases")); + let lease_key = LeaseKey::new("magic-context-kernel", "sqlite", "core"); + let lease = lease_store.acquire(&lease_key).map_err(map_lease_error)?; + let lease_epoch = lease.epoch(); + + if reset_marker_path(&db_path).exists() { + resume_quarantine(&db_path)?; + } + + let expected = expected_identity()?; + let header = inspect_header(&db_path)?; + let mut writer = match header { + HeaderState::Pristine => bootstrap(&db_path)?, + HeaderState::Kernel => { + let mut conn = open_writer(&db_path).map_err(|_| KernelError::Inconclusive)?; + apply_preclassification_profile(&conn).map_err(|_| KernelError::Inconclusive)?; + match classify_open_kernel(&mut conn, &expected)? { + OpenIdentity::Exact => conn, + OpenIdentity::Mismatch { incarnation } => { + drop(conn); + quarantine(&db_path, &incarnation, lease_epoch)?; + bootstrap(&db_path)? + } + } + } + }; + + activate_wal(&writer)?; + stamp_writer_fence(&mut writer, lease_epoch)?; + harden_family(&db_path)?; + let readers = open_read_pool(&db_path)?; + harden_family(&db_path)?; + + Ok(Self { + writer: Mutex::new(writer), + readers, + next_reader: AtomicUsize::new(0), + lease_epoch, + _lease: lease, + }) + } + + pub fn lease_epoch(&self) -> u64 { + self.lease_epoch + } + + #[allow( + dead_code, + reason = "connection ownership is restricted to kernel mutation modules" + )] + pub(crate) fn with_writer( + &self, + operation: impl FnOnce(&mut Connection) -> rusqlite::Result, + ) -> Result { + let mut writer = self.writer.lock().map_err(|_| KernelError::Io)?; + // Fence validation belongs inside the `BEGIN IMMEDIATE` mutation + // transaction, where validation and the write are atomic. + let durable_epoch: i64 = writer + .query_row( + "SELECT writer_epoch FROM writer_fence WHERE id=0", + [], + |row| row.get(0), + ) + .map_err(|_| KernelError::FenceLost)?; + if u64::try_from(durable_epoch).ok() != Some(self.lease_epoch) { + return Err(KernelError::FenceLost); + } + operation(&mut writer).map_err(|_| KernelError::Io) + } + + #[allow( + dead_code, + reason = "connection ownership is restricted to kernel query modules" + )] + pub(crate) fn with_reader( + &self, + operation: impl FnOnce(&Connection) -> rusqlite::Result, + ) -> Result { + let index = self.next_reader.fetch_add(1, Ordering::Relaxed) % self.readers.len(); + let reader = self.readers[index].lock().map_err(|_| KernelError::Io)?; + operation(&reader).map_err(|_| KernelError::Io) + } +} + +enum HeaderState { + Pristine, + Kernel, +} + +fn inspect_header(path: &Path) -> Result { + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + if family_sidecars(path).iter().any(|sidecar| sidecar.exists()) { + return Err(KernelError::Inconclusive); + } + return Ok(HeaderState::Pristine); + } + Err(_) => return Err(KernelError::Io), + }; + if !metadata.is_file() { + return Err(KernelError::Inconclusive); + } + if metadata.len() == 0 { + if family_sidecars(path).iter().any(|sidecar| sidecar.exists()) { + return Err(KernelError::Inconclusive); + } + return Ok(HeaderState::Pristine); + } + if metadata.len() < 100 { + return Err(KernelError::Inconclusive); + } + let mut header = [0_u8; 100]; + File::open(path) + .and_then(|mut file| file.read_exact(&mut header)) + .map_err(|_| KernelError::Inconclusive)?; + if &header[..16] != SQLITE_HEADER { + return Err(KernelError::Foreign); + } + let application_id = + u32::from_be_bytes(header[68..72].try_into().map_err(|_| KernelError::Io)?); + if application_id != KERNEL_APPLICATION_ID { + return Err(KernelError::Foreign); + } + Ok(HeaderState::Kernel) +} + +struct ExpectedIdentity { + digest: String, + inventory: Vec<(String, String)>, +} + +fn expected_identity() -> Result { + let mut conn = Connection::open_in_memory().map_err(|_| KernelError::Io)?; + apply_kernel_schema(&mut conn, "00000000000000000000000000000000", 0) + .map_err(|_| KernelError::Io)?; + Ok(ExpectedIdentity { + digest: kernel_schema_digest(&conn).map_err(|_| KernelError::Io)?, + inventory: kernel_schema_object_inventory(&conn).map_err(|_| KernelError::Io)?, + }) +} + +enum OpenIdentity { + Exact, + Mismatch { incarnation: String }, +} + +struct FormatMarker { + epoch: i64, + incarnation: String, + schema_digest: String, + created_at: i64, + marker_digest: String, +} + +fn classify_open_kernel( + conn: &mut Connection, + expected: &ExpectedIdentity, +) -> Result { + let tx = conn.transaction().map_err(|_| KernelError::Inconclusive)?; + let quick_check: String = tx + .query_row("PRAGMA quick_check(1)", [], |row| row.get(0)) + .map_err(|_| KernelError::Inconclusive)?; + if quick_check != "ok" { + return Err(KernelError::Inconclusive); + } + let marker = read_valid_marker(&tx)?; + let application_id: u32 = tx + .query_row("PRAGMA application_id", [], |row| row.get(0)) + .map_err(|_| KernelError::Inconclusive)?; + let user_version: i64 = tx + .query_row("PRAGMA user_version", [], |row| row.get(0)) + .map_err(|_| KernelError::Inconclusive)?; + let inventory = kernel_schema_object_inventory(&tx).map_err(|_| KernelError::Inconclusive)?; + let digest = kernel_schema_digest(&tx).map_err(|_| KernelError::Inconclusive)?; + let exact = application_id == KERNEL_APPLICATION_ID + && user_version == KERNEL_FORMAT_EPOCH + && marker.epoch == KERNEL_FORMAT_EPOCH + && marker.schema_digest == expected.digest + && digest == expected.digest + && inventory == expected.inventory; + tx.commit().map_err(|_| KernelError::Inconclusive)?; + if exact { + Ok(OpenIdentity::Exact) + } else { + Ok(OpenIdentity::Mismatch { + incarnation: marker.incarnation, + }) + } +} + +fn read_valid_marker(conn: &Connection) -> Result { + let present: Option = conn + .query_row( + "SELECT 1 FROM sqlite_schema WHERE type='table' AND name='mc_kernel_format_marker'", + [], + |row| row.get(0), + ) + .optional() + .map_err(|_| KernelError::Inconclusive)?; + if present.is_none() { + return Err(KernelError::Inconclusive); + } + let mut statement = conn + .prepare( + "SELECT format_epoch,database_incarnation_id,schema_digest,created_at,marker_digest + FROM mc_kernel_format_marker", + ) + .map_err(|_| KernelError::Inconclusive)?; + let rows = statement + .query_map([], |row| { + Ok(FormatMarker { + epoch: row.get(0)?, + incarnation: row.get(1)?, + schema_digest: row.get(2)?, + created_at: row.get(3)?, + marker_digest: row.get(4)?, + }) + }) + .map_err(|_| KernelError::Inconclusive)? + .collect::>>() + .map_err(|_| KernelError::Inconclusive)?; + let [marker] = rows.as_slice() else { + return Err(KernelError::Inconclusive); + }; + if !is_lower_hex(&marker.incarnation, 32) + || !is_lower_hex(&marker.schema_digest, 64) + || !is_lower_hex(&marker.marker_digest, 64) + || marker.epoch < 1 + || marker.created_at < 0 + { + return Err(KernelError::Inconclusive); + } + let expected_marker_digest = compute_marker_digest_for_application_id( + KERNEL_APPLICATION_ID, + marker.epoch, + &marker.incarnation, + &marker.schema_digest, + marker.created_at, + ); + if marker.marker_digest != expected_marker_digest { + return Err(KernelError::Inconclusive); + } + Ok(FormatMarker { + epoch: marker.epoch, + incarnation: marker.incarnation.clone(), + schema_digest: marker.schema_digest.clone(), + created_at: marker.created_at, + marker_digest: marker.marker_digest.clone(), + }) +} + +fn is_lower_hex(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +fn open_writer(path: &Path) -> rusqlite::Result { + Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_WRITE + | OpenFlags::SQLITE_OPEN_CREATE + | OpenFlags::SQLITE_OPEN_NO_MUTEX, + ) +} + +fn bootstrap(path: &Path) -> Result { + let mut conn = open_writer(path).map_err(|_| KernelError::Io)?; + apply_preclassification_profile(&conn).map_err(|_| KernelError::Io)?; + let incarnation: String = conn + .query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0)) + .map_err(|_| KernelError::Io)?; + apply_kernel_schema(&mut conn, &incarnation, current_time_ms()).map_err(|_| KernelError::Io)?; + Ok(conn) +} + +fn apply_preclassification_profile(conn: &Connection) -> rusqlite::Result<()> { + conn.pragma_update(None, "foreign_keys", "ON")?; + conn.pragma_update(None, "trusted_schema", "OFF")?; + conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)?; + Ok(()) +} + +fn activate_wal(conn: &Connection) -> Result<(), KernelError> { + conn.pragma_update(None, "journal_mode", "WAL") + .map_err(|_| KernelError::Io)?; + conn.pragma_update(None, "synchronous", "FULL") + .map_err(|_| KernelError::Io) +} + +fn stamp_writer_fence(conn: &mut Connection, epoch: u64) -> Result<(), KernelError> { + let epoch = i64::try_from(epoch).map_err(|_| KernelError::IdentityMismatch)?; + let tx = conn.transaction().map_err(|_| KernelError::Io)?; + if tx + .execute( + "UPDATE writer_fence SET writer_epoch=?1 WHERE id=0", + [epoch], + ) + .map_err(|_| KernelError::Io)? + != 1 + { + return Err(KernelError::IdentityMismatch); + } + tx.commit().map_err(|_| KernelError::Io) +} + +fn open_read_pool(path: &Path) -> Result>, KernelError> { + (0..READ_POOL_SIZE) + .map(|_| { + let conn = Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, + ) + .map_err(|_| KernelError::Io)?; + conn.pragma_update(None, "query_only", "ON") + .map_err(|_| KernelError::Io)?; + apply_preclassification_profile(&conn).map_err(|_| KernelError::Io)?; + Ok(Mutex::new(conn)) + }) + .collect() +} + +fn family_sidecars(path: &Path) -> [PathBuf; 3] { + [ + suffix_path(path, "-wal"), + suffix_path(path, "-shm"), + suffix_path(path, "-journal"), + ] +} + +fn harden_family(path: &Path) -> Result<(), KernelError> { + protect_file(path).map_err(|_| KernelError::Io)?; + for sidecar in family_sidecars(path) { + protect_file(&sidecar).map_err(|_| KernelError::Io)?; + } + Ok(()) +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct ResetMarker { + protocol: String, + db_path: PathBuf, + database_incarnation_id: String, + quarantine_dir: PathBuf, + marker_digest: String, +} + +fn quarantine(path: &Path, incarnation: &str, lease_epoch: u64) -> Result<(), KernelError> { + let quarantine_dir = allocate_quarantine_dir(path, lease_epoch)?; + let mut marker = ResetMarker { + protocol: RESET_MARKER_PROTOCOL.to_string(), + db_path: path.to_path_buf(), + database_incarnation_id: incarnation.to_string(), + quarantine_dir, + marker_digest: String::new(), + }; + marker.marker_digest = reset_marker_digest(&marker); + publish_reset_marker(path, &marker)?; + move_family(path, &marker) +} + +fn resume_quarantine(path: &Path) -> Result<(), KernelError> { + let bytes = fs::read(reset_marker_path(path)).map_err(|_| KernelError::Inconclusive)?; + let marker: ResetMarker = + serde_json::from_slice(&bytes).map_err(|_| KernelError::Inconclusive)?; + if marker.protocol != RESET_MARKER_PROTOCOL + || marker.db_path != path + || !is_lower_hex(&marker.database_incarnation_id, 32) + || marker.marker_digest != reset_marker_digest(&marker) + || !valid_quarantine_path(path, &marker.quarantine_dir) + { + return Err(KernelError::Inconclusive); + } + move_family(path, &marker) +} + +fn publish_reset_marker(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> { + let marker_path = reset_marker_path(path); + let bytes = serde_json::to_vec(marker).map_err(|_| KernelError::Io)?; + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options.open(&marker_path).map_err(|_| KernelError::Io)?; + if file + .write_all(&bytes) + .and_then(|()| file.sync_all()) + .is_err() + { + drop(file); + let _ = fs::remove_file(&marker_path); + return Err(KernelError::Io); + } + protect_file(&marker_path).map_err(|_| KernelError::Io)?; + sync_parent(path)?; + Ok(()) +} + +fn move_family(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> { + prepare_private_dir(&marker.quarantine_dir)?; + for source in [ + suffix_path(path, "-journal"), + suffix_path(path, "-wal"), + suffix_path(path, "-shm"), + path.to_path_buf(), + ] { + move_one(&source, &marker.quarantine_dir)?; + } + let marker_path = reset_marker_path(path); + move_one(&marker_path, &marker.quarantine_dir)?; + sync_directory(&marker.quarantine_dir)?; + sync_parent(path) +} + +fn move_one(source: &Path, destination_dir: &Path) -> Result<(), KernelError> { + let name = source.file_name().ok_or(KernelError::Inconclusive)?; + let destination = destination_dir.join(name); + let source_exists = source.exists(); + let destination_exists = destination.exists(); + match (source_exists, destination_exists) { + (true, false) => { + fs::rename(source, &destination).map_err(|_| KernelError::Io)?; + protect_file(&destination).map_err(|_| KernelError::Io) + } + (false, true) => protect_file(&destination).map_err(|_| KernelError::Io), + (false, false) => Ok(()), + (true, true) => Err(KernelError::Inconclusive), + } +} + +fn reset_marker_digest(marker: &ResetMarker) -> String { + let canonical = format!( + "{RESET_MARKER_PROTOCOL}\ndb_path={}\ndatabase_incarnation_id={}\nquarantine_dir={}", + marker.db_path.display(), + marker.database_incarnation_id, + marker.quarantine_dir.display() + ); + format!("{:x}", Sha256::digest(canonical.as_bytes())) +} + +fn allocate_quarantine_dir(path: &Path, lease_epoch: u64) -> Result { + let base = PathBuf::from(format!( + "{}{}{}", + path.display(), + QUARANTINE_INFIX, + lease_epoch + )); + for suffix in 0..10_000_u32 { + let candidate = if suffix == 0 { + base.clone() + } else { + PathBuf::from(format!("{}-{suffix}", base.display())) + }; + if !candidate.exists() { + return Ok(candidate); + } + } + Err(KernelError::Io) +} + +fn valid_quarantine_path(path: &Path, quarantine: &Path) -> bool { + quarantine.parent() == path.parent() + && quarantine.file_name().is_some_and(|name| { + name.to_string_lossy().starts_with(&format!( + "{}{}", + path.file_name().unwrap_or_default().to_string_lossy(), + QUARANTINE_INFIX + )) + }) +} + +fn reset_marker_path(path: &Path) -> PathBuf { + PathBuf::from(format!("{}{}", path.display(), RESET_MARKER_SUFFIX)) +} + +fn suffix_path(path: &Path, suffix: &str) -> PathBuf { + PathBuf::from(format!("{}{suffix}", path.display())) +} + +fn absolute_path(path: &Path) -> Result { + if path.is_absolute() { + Ok(path.to_path_buf()) + } else { + std::env::current_dir() + .map(|current| current.join(path)) + .map_err(|_| KernelError::Io) + } +} + +fn prepare_root(root: &Path) -> Result<(), KernelError> { + fs::create_dir_all(root).map_err(|_| KernelError::Io)?; + prepare_private_dir(root) +} + +fn prepare_private_dir(path: &Path) -> Result<(), KernelError> { + if !path.exists() { + fs::create_dir(path).map_err(|_| KernelError::Io)?; + } + let metadata = fs::symlink_metadata(path).map_err(|_| KernelError::Io)?; + if !metadata.is_dir() || metadata.file_type().is_symlink() { + return Err(KernelError::Io); + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(0o700)) + .map_err(|_| KernelError::Io)?; + } + Ok(()) +} + +fn sync_parent(path: &Path) -> Result<(), KernelError> { + let parent = path.parent().ok_or(KernelError::Io)?; + sync_directory(parent) +} + +fn sync_directory(path: &Path) -> Result<(), KernelError> { + File::open(path) + .and_then(|directory| directory.sync_all()) + .map_err(|_| KernelError::Io) +} + +fn current_time_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)) + .unwrap_or(0) +} + +fn map_lease_error(error: LeaseError) -> KernelError { + match error { + LeaseError::Held { .. } => KernelError::Held, + LeaseError::Io(_) => KernelError::Io, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn owned_read_connections_are_query_only() { + let directory = tempfile::tempdir().unwrap(); + let store = KernelStore::open(directory.path()).unwrap(); + store + .with_reader(|connection| { + assert_eq!( + connection.query_row("PRAGMA query_only", [], |row| row.get::<_, i64>(0))?, + 1 + ); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn stale_writer_fence_blocks_the_operation() { + let directory = tempfile::tempdir().unwrap(); + let store = KernelStore::open(directory.path()).unwrap(); + store + .with_writer(|connection| { + connection.execute( + "UPDATE writer_fence SET writer_epoch=writer_epoch+1 WHERE id=0", + [], + )?; + Ok(()) + }) + .unwrap(); + let mut called = false; + let error = store + .with_writer(|_| { + called = true; + Ok(()) + }) + .unwrap_err(); + assert_eq!(error, KernelError::FenceLost); + assert!(!called); + } +} diff --git a/crates/mc-store/src/kernel/schema.rs b/crates/mc-store/src/kernel/schema.rs index 1b0556fd3..43701eecf 100644 --- a/crates/mc-store/src/kernel/schema.rs +++ b/crates/mc-store/src/kernel/schema.rs @@ -1,9 +1,12 @@ -use crate::sqlite_runtime::{DIRECT_FORMAT_EPOCH, MC_APPLICATION_ID}; +use crate::sqlite_runtime::{ + compute_marker_digest_for_application_id, DIRECT_FORMAT_EPOCH, MC_APPLICATION_ID, +}; use rusqlite::{params, Connection, TransactionBehavior}; use sha2::{Digest, Sha256}; use std::collections::BTreeSet; pub const KERNEL_APPLICATION_ID: u32 = MC_APPLICATION_ID; +pub const KERNEL_FORMAT_EPOCH: i64 = DIRECT_FORMAT_EPOCH; pub const KERNEL_SCHEMA_COMPONENT_NAMES: &[&str] = &[ "commit_log", "change_event", @@ -161,7 +164,7 @@ const COMPONENTS: &[(&str, &str)] = &[ ), ( "mc_kernel_format_marker", - r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT,'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT,'mc_kernel_format_marker is immutable'); END;"#, + r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL,marker_digest TEXT NOT NULL CHECK(length(marker_digest)=64)) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END;"#, ), ]; @@ -291,14 +294,21 @@ fn apply_schema rusqlite::Result<()>>( return Err(rusqlite::Error::InvalidQuery); } tx.pragma_update(None, "application_id", KERNEL_APPLICATION_ID)?; - tx.pragma_update(None, "user_version", DIRECT_FORMAT_EPOCH)?; + tx.pragma_update(None, "user_version", KERNEL_FORMAT_EPOCH)?; for (_, sql) in COMPONENTS { tx.execute_batch(sql)?; } tx.execute("INSERT INTO writer_fence(id) VALUES(0)", [])?; hook()?; let digest = kernel_schema_digest(&tx)?; - tx.execute("INSERT INTO mc_kernel_format_marker(singleton,format_epoch,database_incarnation_id,schema_digest,created_at) VALUES(1,1,?1,?2,?3)", params![incarnation,digest,created_at])?; + let marker_digest = compute_marker_digest_for_application_id( + KERNEL_APPLICATION_ID, + KERNEL_FORMAT_EPOCH, + incarnation, + &digest, + created_at, + ); + tx.execute("INSERT INTO mc_kernel_format_marker(singleton,format_epoch,database_incarnation_id,schema_digest,created_at,marker_digest) VALUES(1,?1,?2,?3,?4,?5)", params![KERNEL_FORMAT_EPOCH,incarnation,digest,created_at,marker_digest])?; tx.commit() } @@ -330,6 +340,20 @@ pub fn kernel_schema_inventory(conn: &Connection) -> rusqlite::Result rusqlite::Result> { + let mut stmt = conn.prepare( + "SELECT type,name FROM sqlite_schema + WHERE name NOT LIKE 'sqlite\\_%' ESCAPE '\\' + ORDER BY type,name", + )?; + let inventory = stmt + .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect(); + inventory +} + pub fn kernel_schema_digest(conn: &Connection) -> rusqlite::Result { let mut stmt = conn.prepare("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite\\_%' ESCAPE '\\' AND sql IS NOT NULL ORDER BY type,name")?; let mut rows = stmt.query([])?; diff --git a/crates/mc-store/src/sqlite_runtime.rs b/crates/mc-store/src/sqlite_runtime.rs index dacdf3f34..b853f41ff 100644 --- a/crates/mc-store/src/sqlite_runtime.rs +++ b/crates/mc-store/src/sqlite_runtime.rs @@ -177,10 +177,26 @@ pub fn compute_marker_digest( database_incarnation_id: &str, component_manifest_digest: &str, created_at_ms: i64, +) -> String { + compute_marker_digest_for_application_id( + MC_APPLICATION_ID, + format_epoch, + database_incarnation_id, + component_manifest_digest, + created_at_ms, + ) +} + +pub fn compute_marker_digest_for_application_id( + application_id: u32, + format_epoch: i64, + database_incarnation_id: &str, + component_manifest_digest: &str, + created_at_ms: i64, ) -> String { let lines = [ FORMAT_MARKER_DIGEST_PROTOCOL.to_string(), - format!("application_id={MC_APPLICATION_ID}"), + format!("application_id={application_id}"), format!("format_epoch={format_epoch}"), format!("database_incarnation_id={database_incarnation_id}"), format!("component_manifest_digest={component_manifest_digest}"), diff --git a/crates/mc-store/tests/kernel_open.rs b/crates/mc-store/tests/kernel_open.rs new file mode 100644 index 000000000..64dd8b93f --- /dev/null +++ b/crates/mc-store/tests/kernel_open.rs @@ -0,0 +1,315 @@ +use mc_store::kernel::schema::{ + apply_kernel_connection_profile, apply_kernel_schema, kernel_schema_digest, + KERNEL_APPLICATION_ID, +}; +use mc_store::kernel::{KernelError, KernelStore}; +use mc_store::sqlite_runtime::compute_marker_digest_for_application_id; +#[cfg(feature = "test-support")] +use mc_store::sqlite_runtime::SqliteEngineIdentity; +use rusqlite::{Connection, OpenFlags}; +use sha2::{Digest, Sha256}; +use std::fs; +use std::path::{Path, PathBuf}; + +const INCARNATION: &str = "0123456789abcdef0123456789abcdef"; + +fn core_path(root: &Path) -> PathBuf { + root.join("core.sqlite") +} + +fn inspect(root: &Path, query: impl FnOnce(&Connection) -> T) -> T { + let conn = Connection::open_with_flags(core_path(root), OpenFlags::SQLITE_OPEN_READ_ONLY) + .expect("open read-only inspection connection"); + query(&conn) +} + +fn seed_kernel(root: &Path) -> Connection { + fs::create_dir_all(root).unwrap(); + let mut conn = Connection::open(core_path(root)).unwrap(); + apply_kernel_connection_profile(&mut conn, 5_000).unwrap(); + apply_kernel_schema(&mut conn, INCARNATION, 1_000).unwrap(); + conn +} + +fn marker_digest(epoch: i64, schema_digest: &str) -> String { + compute_marker_digest_for_application_id( + KERNEL_APPLICATION_ID, + epoch, + INCARNATION, + schema_digest, + 1_000, + ) +} + +fn replace_marker(conn: &Connection, epoch: i64, schema_digest: &str) { + conn.execute_batch( + "DROP TRIGGER mc_kernel_format_marker_no_update; + DROP TRIGGER mc_kernel_format_marker_no_delete;", + ) + .unwrap(); + conn.execute( + "UPDATE mc_kernel_format_marker + SET format_epoch=?1, schema_digest=?2, marker_digest=?3", + (epoch, schema_digest, marker_digest(epoch, schema_digest)), + ) + .unwrap(); + conn.execute_batch( + "CREATE TRIGGER mc_kernel_format_marker_no_update + BEFORE UPDATE ON mc_kernel_format_marker BEGIN + SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); + END; + CREATE TRIGGER mc_kernel_format_marker_no_delete + BEFORE DELETE ON mc_kernel_format_marker BEGIN + SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); + END;", + ) + .unwrap(); +} + +fn quarantine_dirs(root: &Path) -> Vec { + let mut paths = fs::read_dir(root) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| { + path.file_name() + .unwrap() + .to_string_lossy() + .starts_with("core.sqlite.mc-quarantine-") + }) + .collect::>(); + paths.sort(); + paths +} + +#[test] +fn fresh_open_and_exact_reopen_preserve_identity_and_advance_fence() { + let dir = tempfile::tempdir().unwrap(); + let first = KernelStore::open(dir.path()).unwrap(); + let first_epoch = first.lease_epoch(); + let incarnation = inspect(dir.path(), |conn| { + conn.query_row( + "SELECT database_incarnation_id FROM mc_kernel_format_marker", + [], + |row| row.get::<_, String>(0), + ) + }) + .unwrap(); + assert_eq!( + inspect(dir.path(), |conn| conn.query_row( + "SELECT writer_epoch FROM writer_fence WHERE id=0", + [], + |row| row.get::<_, i64>(0) + )) + .unwrap(), + i64::try_from(first_epoch).unwrap() + ); + drop(first); + + let second = KernelStore::open(dir.path()).unwrap(); + assert!(second.lease_epoch() > first_epoch); + assert_eq!( + inspect(dir.path(), |conn| { + conn.query_row( + "SELECT database_incarnation_id FROM mc_kernel_format_marker", + [], + |row| row.get::<_, String>(0), + ) + }) + .unwrap(), + incarnation + ); +} + +#[test] +fn second_opener_is_held_without_touching_database_family() { + let dir = tempfile::tempdir().unwrap(); + let first = KernelStore::open(dir.path()).unwrap(); + let before = fs::read(core_path(dir.path())).unwrap(); + assert_eq!( + KernelStore::open(dir.path()).unwrap_err(), + KernelError::Held + ); + assert_eq!(fs::read(core_path(dir.path())).unwrap(), before); + drop(first); +} + +#[test] +fn every_conclusive_kernel_mismatch_is_quarantined_and_rebuilt() { + for mismatch in ["epoch", "digest", "inventory"] { + let dir = tempfile::tempdir().unwrap(); + let conn = seed_kernel(dir.path()); + match mismatch { + "epoch" => { + let digest = kernel_schema_digest(&conn).unwrap(); + replace_marker(&conn, 2, &digest); + } + "digest" => replace_marker(&conn, 1, &"a".repeat(64)), + "inventory" => { + conn.execute_batch("CREATE TABLE unexpected(value INTEGER) STRICT;") + .unwrap(); + } + _ => unreachable!(), + } + drop(conn); + + let _store = KernelStore::open(dir.path()).unwrap(); + assert_eq!(quarantine_dirs(dir.path()).len(), 1, "{mismatch}"); + assert_eq!( + inspect(dir.path(), |conn| conn.query_row( + "SELECT COUNT(*) FROM mc_kernel_format_marker", + [], + |row| row.get::<_, i64>(0) + )) + .unwrap(), + 1 + ); + } +} + +#[test] +fn foreign_family_is_refused_before_sqlite_can_touch_it() { + let dir = tempfile::tempdir().unwrap(); + let path = core_path(dir.path()); + let conn = Connection::open(&path).unwrap(); + conn.pragma_update(None, "application_id", 0x4D43_5458_u32) + .unwrap(); + conn.execute_batch("CREATE TABLE legacy(value TEXT);") + .unwrap(); + drop(conn); + fs::write(format!("{}-wal", path.display()), b"foreign wal").unwrap(); + let before_main = fs::read(&path).unwrap(); + let before_wal = fs::read(format!("{}-wal", path.display())).unwrap(); + + assert_eq!( + KernelStore::open(dir.path()).unwrap_err(), + KernelError::Foreign + ); + assert_eq!(fs::read(&path).unwrap(), before_main); + assert_eq!( + fs::read(format!("{}-wal", path.display())).unwrap(), + before_wal + ); + assert!(quarantine_dirs(dir.path()).is_empty()); +} + +#[test] +fn malformed_marker_is_inconclusive_and_untouched() { + let dir = tempfile::tempdir().unwrap(); + let conn = seed_kernel(dir.path()); + conn.execute_batch("DROP TRIGGER mc_kernel_format_marker_no_update;") + .unwrap(); + conn.execute( + "UPDATE mc_kernel_format_marker SET marker_digest=?1", + ["g".repeat(64)], + ) + .unwrap(); + drop(conn); + let path = core_path(dir.path()); + let before = fs::read(&path).unwrap(); + + assert_eq!( + KernelStore::open(dir.path()).unwrap_err(), + KernelError::Inconclusive + ); + assert_eq!(fs::read(&path).unwrap(), before); + assert!(quarantine_dirs(dir.path()).is_empty()); +} + +#[test] +fn valid_interrupted_reset_marker_resumes_without_opening_old_family() { + let dir = tempfile::tempdir().unwrap(); + let conn = seed_kernel(dir.path()); + conn.execute_batch("CREATE TABLE unexpected(value INTEGER) STRICT;") + .unwrap(); + drop(conn); + let db_path = core_path(dir.path()).canonicalize().unwrap(); + let quarantine = dir.path().join("core.sqlite.mc-quarantine-resume"); + fs::create_dir(&quarantine).unwrap(); + fs::rename(&db_path, quarantine.join("core.sqlite")).unwrap(); + for suffix in ["-journal", "-wal", "-shm"] { + fs::write(format!("{}{suffix}", db_path.display()), suffix.as_bytes()).unwrap(); + } + let marker_without_digest = serde_json::json!({ + "protocol": "mc-kernel-reset-marker-v1", + "db_path": db_path, + "database_incarnation_id": INCARNATION, + "quarantine_dir": quarantine, + }); + let canonical = format!( + "mc-kernel-reset-marker-v1\ndb_path={}\ndatabase_incarnation_id={}\nquarantine_dir={}", + marker_without_digest["db_path"].as_str().unwrap(), + INCARNATION, + marker_without_digest["quarantine_dir"].as_str().unwrap(), + ); + let digest = format!("{:x}", Sha256::digest(canonical.as_bytes())); + let mut marker = marker_without_digest; + marker["marker_digest"] = digest.into(); + fs::write( + dir.path().join("core.sqlite.mc-reset"), + serde_json::to_vec(&marker).unwrap(), + ) + .unwrap(); + + let _store = KernelStore::open(dir.path()).unwrap(); + assert!(core_path(dir.path()).is_file()); + assert!(quarantine.join("core.sqlite.mc-reset").is_file()); + assert_owner_only(&quarantine, 0o700); + for name in [ + "core.sqlite", + "core.sqlite-journal", + "core.sqlite-wal", + "core.sqlite-shm", + "core.sqlite.mc-reset", + ] { + assert_owner_only(&quarantine.join(name), 0o600); + } +} + +#[test] +fn writer_and_sidecars_are_owner_only() { + let dir = tempfile::tempdir().unwrap(); + let store = KernelStore::open(dir.path()).unwrap(); + let path = core_path(dir.path()); + assert_owner_only(&path, 0o600); + assert_owner_only(&PathBuf::from(format!("{}-wal", path.display())), 0o600); + assert_owner_only(&PathBuf::from(format!("{}-shm", path.display())), 0o600); + assert_eq!( + inspect(dir.path(), |conn| conn.query_row( + "PRAGMA journal_mode", + [], + |row| row.get::<_, String>(0) + )) + .unwrap(), + "wal" + ); + drop(store); +} + +#[cfg(feature = "test-support")] +#[test] +fn unsupported_engine_is_rejected_before_creating_lease_or_database_files() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("kernel"); + let identity = SqliteEngineIdentity { + sqlite_version: "3.51.2".to_string(), + sqlite_source_id: "2026-01-01 00:00:00 0123456789abcdef0123456789abcdef01234567" + .to_string(), + }; + assert_eq!( + KernelStore::open_with_engine_identity_for_test(&root, &identity).unwrap_err(), + KernelError::EngineUnsupported + ); + assert!(!root.exists()); +} + +#[cfg(unix)] +fn assert_owner_only(path: &Path, expected: u32) { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + fs::metadata(path).unwrap().permissions().mode() & 0o777, + expected + ); +} + +#[cfg(not(unix))] +fn assert_owner_only(_path: &Path, _expected: u32) {} From 2cd5c3a31ff3fde924c214c6f876b07ff59773ba Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 21:06:51 +0000 Subject: [PATCH 2/6] fix(store): harden kernel open against crash and fence races The writer fence was read in autocommit, so a lease successor could stamp a new epoch between the check and the write. Reading it inside a BEGIN IMMEDIATE transaction makes the check and the mutation atomic. Classification opened the family read-write, which lets SQLite roll back a journal or checkpoint a WAL. That rewrites the main file and unlinks its sidecars before the identity is known, breaking the promise that Foreign and Inconclusive leave the family untouched. A read-only connection keeps that promise. Quarantine removed the reset marker before the renames it describes were durable. Since the marker's absence is what declares the reset complete, a crash in that window could lose the renames and leave nothing to recover from. Both directories are now durable while the marker is still live, and the marker is published under a staging name so an interrupted publication cannot leave a marker recovery can only refuse. Derived paths went through Path::display, which replaces non-UTF-8 bytes and named a different file, so the real sidecars were neither hardened nor moved. Suffixes are now appended to the path's OsStr, and the root is canonicalized so recovery compares one spelling of the directory. Existence checks went through Path::exists, which maps every error to false and let a transient I/O error select a destructive branch. Close remaining recovery hazards: validate journal mode, bound marker reads, give readers a snapshot, and avoid INSERT OR REPLACE, which bypasses the format-marker delete trigger. --- crates/mc-store/src/kernel/open.rs | 321 ++++++++++++++++--------- crates/mc-store/src/kernel/schema.rs | 2 +- crates/mc-store/src/lib.rs | 2 +- crates/mc-store/tests/kernel_open.rs | 70 ++++-- crates/mc-store/tests/kernel_schema.rs | 24 +- 5 files changed, 278 insertions(+), 141 deletions(-) diff --git a/crates/mc-store/src/kernel/open.rs b/crates/mc-store/src/kernel/open.rs index b50ee99a6..9e4681ff0 100644 --- a/crates/mc-store/src/kernel/open.rs +++ b/crates/mc-store/src/kernel/open.rs @@ -1,20 +1,22 @@ use cortexkit_lease::{ protect_file, FileLeaseStore, LeaseError, LeaseHandle, LeaseKey, LeaseStore, }; -use rusqlite::{Connection, OpenFlags, OptionalExtension}; +use mc_core::claim_operation::is_lower_hex; +use rusqlite::{Connection, OpenFlags, OptionalExtension, Transaction, TransactionBehavior}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::fmt; use std::fs::{self, File, OpenOptions}; -use std::io::{Read, Write}; +use std::io::{ErrorKind, Read, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Mutex; +use std::sync::{LazyLock, Mutex, PoisonError}; use super::schema::{ apply_kernel_schema, kernel_schema_digest, kernel_schema_object_inventory, KERNEL_APPLICATION_ID, KERNEL_FORMAT_EPOCH, }; +use crate::current_time_ms; use crate::sqlite_runtime::{ compute_marker_digest_for_application_id, evaluate_sqlite_runtime_gate, probe_sqlite_engine_identity_off_path, SqliteEngineIdentity, @@ -24,9 +26,14 @@ const BUSY_TIMEOUT_MS: i64 = 5_000; const READ_POOL_SIZE: usize = 2; const RESET_MARKER_PROTOCOL: &str = "mc-kernel-reset-marker-v1"; const RESET_MARKER_SUFFIX: &str = ".mc-reset"; +const RESET_MARKER_STAGING_SUFFIX: &str = ".staging"; const QUARANTINE_INFIX: &str = ".mc-quarantine-"; const SQLITE_HEADER: &[u8; 16] = b"SQLite format 3\0"; +/// Limit marker reads to 64 KiB so invalid marker data cannot control +/// allocation size. +const RESET_MARKER_MAX_BYTES: u64 = 64 * 1024; + #[derive(Clone, Copy, PartialEq, Eq)] pub enum KernelError { Held, @@ -103,38 +110,41 @@ impl KernelStore { } fn open_supported(root: impl AsRef) -> Result { - let root = absolute_path(root.as_ref())?; - prepare_root(&root)?; + let root = prepare_root(root.as_ref())?; let db_path = root.join("core.sqlite"); let lease_store = FileLeaseStore::new(root.join("leases")); let lease_key = LeaseKey::new("magic-context-kernel", "sqlite", "core"); let lease = lease_store.acquire(&lease_key).map_err(map_lease_error)?; let lease_epoch = lease.epoch(); - if reset_marker_path(&db_path).exists() { + if entry_exists(&reset_marker_path(&db_path))? { resume_quarantine(&db_path)?; } - let expected = expected_identity()?; let header = inspect_header(&db_path)?; let mut writer = match header { HeaderState::Pristine => bootstrap(&db_path)?, - HeaderState::Kernel => { - let mut conn = open_writer(&db_path).map_err(|_| KernelError::Inconclusive)?; - apply_preclassification_profile(&conn).map_err(|_| KernelError::Inconclusive)?; - match classify_open_kernel(&mut conn, &expected)? { - OpenIdentity::Exact => conn, - OpenIdentity::Mismatch { incarnation } => { - drop(conn); - quarantine(&db_path, &incarnation, lease_epoch)?; - bootstrap(&db_path)? - } + HeaderState::Kernel => match classify_existing_family(&db_path)? { + OpenIdentity::Exact => { + let conn = open_writer(&db_path).map_err(|_| KernelError::Inconclusive)?; + apply_preclassification_profile(&conn) + .map_err(|_| KernelError::Inconclusive)?; + conn + } + OpenIdentity::Mismatch { incarnation } => { + quarantine(&db_path, &incarnation, lease_epoch)?; + bootstrap(&db_path)? } - } + }, }; activate_wal(&writer)?; stamp_writer_fence(&mut writer, lease_epoch)?; + // Two hardening passes are required because the family grows between + // them. WAL activation creates `-wal` and `-shm` under the process umask, + // so the first pass restricts them as early as possible; a read-only open + // recreates `-shm` when it is missing, so the second pass restricts that + // one too. harden_family(&db_path)?; let readers = open_read_pool(&db_path)?; harden_family(&db_path)?; @@ -158,12 +168,15 @@ impl KernelStore { )] pub(crate) fn with_writer( &self, - operation: impl FnOnce(&mut Connection) -> rusqlite::Result, + operation: impl FnOnce(&Transaction<'_>) -> rusqlite::Result, ) -> Result { - let mut writer = self.writer.lock().map_err(|_| KernelError::Io)?; - // Fence validation belongs inside the `BEGIN IMMEDIATE` mutation - // transaction, where validation and the write are atomic. - let durable_epoch: i64 = writer + let mut writer = self.writer.lock().unwrap_or_else(PoisonError::into_inner); + // `BEGIN IMMEDIATE` blocks a competing writer until this transaction ends. + // The fence check and the mutation are therefore atomic. + let tx = writer + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|_| KernelError::Io)?; + let durable_epoch: i64 = tx .query_row( "SELECT writer_epoch FROM writer_fence WHERE id=0", [], @@ -173,7 +186,9 @@ impl KernelStore { if u64::try_from(durable_epoch).ok() != Some(self.lease_epoch) { return Err(KernelError::FenceLost); } - operation(&mut writer).map_err(|_| KernelError::Io) + let value = operation(&tx).map_err(|_| KernelError::Io)?; + tx.commit().map_err(|_| KernelError::Io)?; + Ok(value) } #[allow( @@ -185,8 +200,17 @@ impl KernelStore { operation: impl FnOnce(&Connection) -> rusqlite::Result, ) -> Result { let index = self.next_reader.fetch_add(1, Ordering::Relaxed) % self.readers.len(); - let reader = self.readers[index].lock().map_err(|_| KernelError::Io)?; - operation(&reader).map_err(|_| KernelError::Io) + // Recovering a poisoned reader guard keeps its pool slot usable. + // `Transaction::Drop` rolls back after a closure panic. + let mut reader = self.readers[index] + .lock() + .unwrap_or_else(PoisonError::into_inner); + // After its first read, the transaction keeps one snapshot for the + // closure. Several reads therefore observe one consistent state. + let tx = reader.transaction().map_err(|_| KernelError::Io)?; + let value = operation(&tx).map_err(|_| KernelError::Io)?; + tx.commit().map_err(|_| KernelError::Io)?; + Ok(value) } } @@ -195,25 +219,36 @@ enum HeaderState { Kernel, } +/// `Path::exists` maps every error to `false`; only `NotFound` counts as absence. +fn entry_exists(path: &Path) -> Result { + match fs::symlink_metadata(path) { + Ok(_) => Ok(true), + Err(error) if error.kind() == ErrorKind::NotFound => Ok(false), + Err(_) => Err(KernelError::Io), + } +} + +/// A zero-length main file has never held a committed page. +/// +/// A `-journal` does not prevent treating an empty main file as pristine. +fn classify_empty_family(path: &Path) -> Result { + if entry_exists(&suffix_path(path, "-wal"))? || entry_exists(&suffix_path(path, "-shm"))? { + return Err(KernelError::Inconclusive); + } + Ok(HeaderState::Pristine) +} + fn inspect_header(path: &Path) -> Result { let metadata = match fs::symlink_metadata(path) { Ok(metadata) => metadata, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - if family_sidecars(path).iter().any(|sidecar| sidecar.exists()) { - return Err(KernelError::Inconclusive); - } - return Ok(HeaderState::Pristine); - } + Err(error) if error.kind() == ErrorKind::NotFound => return classify_empty_family(path), Err(_) => return Err(KernelError::Io), }; if !metadata.is_file() { return Err(KernelError::Inconclusive); } if metadata.len() == 0 { - if family_sidecars(path).iter().any(|sidecar| sidecar.exists()) { - return Err(KernelError::Inconclusive); - } - return Ok(HeaderState::Pristine); + return classify_empty_family(path); } if metadata.len() < 100 { return Err(KernelError::Inconclusive); @@ -238,14 +273,37 @@ struct ExpectedIdentity { inventory: Vec<(String, String)>, } -fn expected_identity() -> Result { - let mut conn = Connection::open_in_memory().map_err(|_| KernelError::Io)?; - apply_kernel_schema(&mut conn, "00000000000000000000000000000000", 0) - .map_err(|_| KernelError::Io)?; - Ok(ExpectedIdentity { - digest: kernel_schema_digest(&conn).map_err(|_| KernelError::Io)?, - inventory: kernel_schema_object_inventory(&conn).map_err(|_| KernelError::Io)?, +static EXPECTED_IDENTITY: LazyLock> = LazyLock::new(|| { + let mut conn = Connection::open_in_memory().ok()?; + apply_kernel_schema(&mut conn, "00000000000000000000000000000000", 0).ok()?; + Some(ExpectedIdentity { + digest: kernel_schema_digest(&conn).ok()?, + inventory: kernel_schema_object_inventory(&conn).ok()?, }) +}); + +fn expected_identity() -> Result<&'static ExpectedIdentity, KernelError> { + EXPECTED_IDENTITY.as_ref().ok_or(KernelError::Io) +} + +/// Classification must leave the family byte-identical. +/// +/// The `Foreign` and `Inconclusive` outcomes promise an untouched family. +/// Journal recovery and WAL checkpointing would both break that promise. +fn classify_existing_family(path: &Path) -> Result { + let expected = expected_identity()?; + let mut conn = Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, + ) + .map_err(|_| KernelError::Inconclusive)?; + conn.pragma_update(None, "query_only", "ON") + .map_err(|_| KernelError::Inconclusive)?; + conn.pragma_update(None, "trusted_schema", "OFF") + .map_err(|_| KernelError::Inconclusive)?; + conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS) + .map_err(|_| KernelError::Inconclusive)?; + classify_open_kernel(&mut conn, expected) } enum OpenIdentity { @@ -358,13 +416,6 @@ fn read_valid_marker(conn: &Connection) -> Result { }) } -fn is_lower_hex(value: &str, length: usize) -> bool { - value.len() == length - && value - .bytes() - .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) -} - fn open_writer(path: &Path) -> rusqlite::Result { Connection::open_with_flags( path, @@ -391,9 +442,16 @@ fn apply_preclassification_profile(conn: &Connection) -> rusqlite::Result<()> { Ok(()) } +/// `PRAGMA journal_mode` returns the mode now in effect rather than failing. +/// +/// Readers depend on WAL for snapshot isolation, so the returned mode is checked. fn activate_wal(conn: &Connection) -> Result<(), KernelError> { - conn.pragma_update(None, "journal_mode", "WAL") + let mode: String = conn + .query_row("PRAGMA journal_mode=WAL", [], |row| row.get(0)) .map_err(|_| KernelError::Io)?; + if !mode.eq_ignore_ascii_case("wal") { + return Err(KernelError::Io); + } conn.pragma_update(None, "synchronous", "FULL") .map_err(|_| KernelError::Io) } @@ -471,7 +529,12 @@ fn quarantine(path: &Path, incarnation: &str, lease_epoch: u64) -> Result<(), Ke } fn resume_quarantine(path: &Path) -> Result<(), KernelError> { - let bytes = fs::read(reset_marker_path(path)).map_err(|_| KernelError::Inconclusive)?; + let marker_path = reset_marker_path(path); + let metadata = fs::symlink_metadata(&marker_path).map_err(|_| KernelError::Inconclusive)?; + if !metadata.is_file() || metadata.len() > RESET_MARKER_MAX_BYTES { + return Err(KernelError::Inconclusive); + } + let bytes = fs::read(&marker_path).map_err(|_| KernelError::Inconclusive)?; let marker: ResetMarker = serde_json::from_slice(&bytes).map_err(|_| KernelError::Inconclusive)?; if marker.protocol != RESET_MARKER_PROTOCOL @@ -485,43 +548,51 @@ fn resume_quarantine(path: &Path) -> Result<(), KernelError> { move_family(path, &marker) } +/// The live marker name appears only once its bytes are durable. fn publish_reset_marker(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> { let marker_path = reset_marker_path(path); + let staging = suffix_path(&marker_path, RESET_MARKER_STAGING_SUFFIX); let bytes = serde_json::to_vec(marker).map_err(|_| KernelError::Io)?; + write_private_file(&staging, &bytes)?; + fs::rename(&staging, &marker_path).map_err(|_| KernelError::Io)?; + protect_file(&marker_path).map_err(|_| KernelError::Io)?; + sync_parent(path)?; + Ok(()) +} + +fn write_private_file(path: &Path, bytes: &[u8]) -> Result<(), KernelError> { let mut options = OpenOptions::new(); - options.write(true).create_new(true); + options.write(true).create(true).truncate(true); #[cfg(unix)] { use std::os::unix::fs::OpenOptionsExt; options.mode(0o600); } - let mut file = options.open(&marker_path).map_err(|_| KernelError::Io)?; - if file - .write_all(&bytes) - .and_then(|()| file.sync_all()) - .is_err() - { - drop(file); - let _ = fs::remove_file(&marker_path); + let outcome = options.open(path).and_then(|mut file| { + file.write_all(bytes)?; + file.sync_all() + }); + if outcome.is_err() { + let _ = fs::remove_file(path); return Err(KernelError::Io); } - protect_file(&marker_path).map_err(|_| KernelError::Io)?; - sync_parent(path)?; Ok(()) } +/// The marker's absence declares the reset complete. +/// +/// Move the marker only after both directories are durable. fn move_family(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> { prepare_private_dir(&marker.quarantine_dir)?; - for source in [ - suffix_path(path, "-journal"), - suffix_path(path, "-wal"), - suffix_path(path, "-shm"), - path.to_path_buf(), - ] { + for source in family_sidecars(path) + .into_iter() + .chain([path.to_path_buf()]) + { move_one(&source, &marker.quarantine_dir)?; } - let marker_path = reset_marker_path(path); - move_one(&marker_path, &marker.quarantine_dir)?; + sync_directory(&marker.quarantine_dir)?; + sync_parent(path)?; + move_one(&reset_marker_path(path), &marker.quarantine_dir)?; sync_directory(&marker.quarantine_dir)?; sync_parent(path) } @@ -529,9 +600,7 @@ fn move_family(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> { fn move_one(source: &Path, destination_dir: &Path) -> Result<(), KernelError> { let name = source.file_name().ok_or(KernelError::Inconclusive)?; let destination = destination_dir.join(name); - let source_exists = source.exists(); - let destination_exists = destination.exists(); - match (source_exists, destination_exists) { + match (entry_exists(source)?, entry_exists(&destination)?) { (true, false) => { fs::rename(source, &destination).map_err(|_| KernelError::Io)?; protect_file(&destination).map_err(|_| KernelError::Io) @@ -553,19 +622,14 @@ fn reset_marker_digest(marker: &ResetMarker) -> String { } fn allocate_quarantine_dir(path: &Path, lease_epoch: u64) -> Result { - let base = PathBuf::from(format!( - "{}{}{}", - path.display(), - QUARANTINE_INFIX, - lease_epoch - )); + let base = suffix_path(path, &format!("{QUARANTINE_INFIX}{lease_epoch}")); for suffix in 0..10_000_u32 { let candidate = if suffix == 0 { base.clone() } else { - PathBuf::from(format!("{}-{suffix}", base.display())) + suffix_path(&base, &format!("-{suffix}")) }; - if !candidate.exists() { + if !entry_exists(&candidate)? { return Ok(candidate); } } @@ -573,52 +637,63 @@ fn allocate_quarantine_dir(path: &Path, lease_epoch: u64) -> Result bool { + let Some(file_name) = path.file_name() else { + return false; + }; + let mut prefix = file_name.to_os_string(); + prefix.push(QUARANTINE_INFIX); quarantine.parent() == path.parent() && quarantine.file_name().is_some_and(|name| { - name.to_string_lossy().starts_with(&format!( - "{}{}", - path.file_name().unwrap_or_default().to_string_lossy(), - QUARANTINE_INFIX - )) + name.as_encoded_bytes() + .starts_with(prefix.as_encoded_bytes()) }) } fn reset_marker_path(path: &Path) -> PathBuf { - PathBuf::from(format!("{}{}", path.display(), RESET_MARKER_SUFFIX)) + suffix_path(path, RESET_MARKER_SUFFIX) } +/// `Path::display` replaces non-UTF-8 bytes, which would name a different file. fn suffix_path(path: &Path, suffix: &str) -> PathBuf { - PathBuf::from(format!("{}{suffix}", path.display())) -} - -fn absolute_path(path: &Path) -> Result { - if path.is_absolute() { - Ok(path.to_path_buf()) - } else { - std::env::current_dir() - .map(|current| current.join(path)) - .map_err(|_| KernelError::Io) - } + let mut name = path.as_os_str().to_os_string(); + name.push(suffix); + PathBuf::from(name) } -fn prepare_root(root: &Path) -> Result<(), KernelError> { +/// `resume_quarantine` compares the marker's `db_path` against this path by value. +/// +/// Canonicalizing after creation gives every spelling of one directory one name. +fn prepare_root(root: &Path) -> Result { fs::create_dir_all(root).map_err(|_| KernelError::Io)?; - prepare_private_dir(root) + prepare_private_dir(root)?; + fs::canonicalize(root).map_err(|_| KernelError::Io) } fn prepare_private_dir(path: &Path) -> Result<(), KernelError> { - if !path.exists() { - fs::create_dir(path).map_err(|_| KernelError::Io)?; + // The umask would leave a readable window between `create_dir` and `chmod`. + #[cfg(unix)] + let created = { + use std::os::unix::fs::DirBuilderExt; + fs::DirBuilder::new().mode(0o700).create(path) + }; + #[cfg(not(unix))] + let created = fs::create_dir(path); + match created { + Ok(()) => {} + Err(error) if error.kind() == ErrorKind::AlreadyExists => {} + Err(_) => return Err(KernelError::Io), } let metadata = fs::symlink_metadata(path).map_err(|_| KernelError::Io)?; - if !metadata.is_dir() || metadata.file_type().is_symlink() { + if !metadata.is_dir() { return Err(KernelError::Io); } #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; - fs::set_permissions(path, fs::Permissions::from_mode(0o700)) - .map_err(|_| KernelError::Io)?; + if metadata.permissions().mode() & 0o777 != 0o700 { + fs::set_permissions(path, fs::Permissions::from_mode(0o700)) + .map_err(|_| KernelError::Io)?; + } } Ok(()) } @@ -634,13 +709,6 @@ fn sync_directory(path: &Path) -> Result<(), KernelError> { .map_err(|_| KernelError::Io) } -fn current_time_ms() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)) - .unwrap_or(0) -} - fn map_lease_error(error: LeaseError) -> KernelError { match error { LeaseError::Held { .. } => KernelError::Held, @@ -690,4 +758,31 @@ mod tests { assert_eq!(error, KernelError::FenceLost); assert!(!called); } + + #[test] + fn failed_writer_operation_leaves_no_partial_write() { + let directory = tempfile::tempdir().unwrap(); + let store = KernelStore::open(directory.path()).unwrap(); + let error = store + .with_writer(|tx| -> rusqlite::Result<()> { + tx.execute( + "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause) + VALUES('t1',1,1,'actor','cause')", + [], + )?; + Err(rusqlite::Error::InvalidQuery) + }) + .unwrap_err(); + assert_eq!(error, KernelError::Io); + store + .with_reader(|tx| { + assert_eq!( + tx.query_row("SELECT COUNT(*) FROM commit_log", [], |row| row + .get::<_, i64>(0))?, + 0 + ); + Ok(()) + }) + .unwrap(); + } } diff --git a/crates/mc-store/src/kernel/schema.rs b/crates/mc-store/src/kernel/schema.rs index 43701eecf..31c3dad9f 100644 --- a/crates/mc-store/src/kernel/schema.rs +++ b/crates/mc-store/src/kernel/schema.rs @@ -164,7 +164,7 @@ const COMPONENTS: &[(&str, &str)] = &[ ), ( "mc_kernel_format_marker", - r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL,marker_digest TEXT NOT NULL CHECK(length(marker_digest)=64)) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END;"#, + r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL,marker_digest TEXT NOT NULL CHECK(length(marker_digest)=64)) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_replace BEFORE INSERT ON mc_kernel_format_marker WHEN EXISTS(SELECT 1 FROM mc_kernel_format_marker) BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END;"#, ), ]; diff --git a/crates/mc-store/src/lib.rs b/crates/mc-store/src/lib.rs index fd668d073..c412b532f 100644 --- a/crates/mc-store/src/lib.rs +++ b/crates/mc-store/src/lib.rs @@ -423,7 +423,7 @@ pub const PASS_SCHEDULER_TELEMETRY_MAX_BYTES: usize = 1 + PASS_SCHEDULER_INTERESTING_HISTORY_CAP * (MAX_INTERESTING_PASS_SCHEDULER_OBSERVATION_JSON_BYTES + 1); -fn current_time_ms() -> i64 { +pub(crate) fn current_time_ms() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)) diff --git a/crates/mc-store/tests/kernel_open.rs b/crates/mc-store/tests/kernel_open.rs index 64dd8b93f..1a5af53b5 100644 --- a/crates/mc-store/tests/kernel_open.rs +++ b/crates/mc-store/tests/kernel_open.rs @@ -194,25 +194,29 @@ fn foreign_family_is_refused_before_sqlite_can_touch_it() { #[test] fn malformed_marker_is_inconclusive_and_untouched() { - let dir = tempfile::tempdir().unwrap(); - let conn = seed_kernel(dir.path()); - conn.execute_batch("DROP TRIGGER mc_kernel_format_marker_no_update;") + // "g" fails the lowercase-hex check; the all-zero digest fails digest comparison. + for digest in ["g".repeat(64), "0".repeat(64)] { + let dir = tempfile::tempdir().unwrap(); + let conn = seed_kernel(dir.path()); + conn.execute_batch("DROP TRIGGER mc_kernel_format_marker_no_update;") + .unwrap(); + conn.execute( + "UPDATE mc_kernel_format_marker SET marker_digest=?1", + [&digest], + ) .unwrap(); - conn.execute( - "UPDATE mc_kernel_format_marker SET marker_digest=?1", - ["g".repeat(64)], - ) - .unwrap(); - drop(conn); - let path = core_path(dir.path()); - let before = fs::read(&path).unwrap(); + drop(conn); + let path = core_path(dir.path()); + let before = fs::read(&path).unwrap(); - assert_eq!( - KernelStore::open(dir.path()).unwrap_err(), - KernelError::Inconclusive - ); - assert_eq!(fs::read(&path).unwrap(), before); - assert!(quarantine_dirs(dir.path()).is_empty()); + assert_eq!( + KernelStore::open(dir.path()).unwrap_err(), + KernelError::Inconclusive, + "{digest}" + ); + assert_eq!(fs::read(&path).unwrap(), before, "{digest}"); + assert!(quarantine_dirs(dir.path()).is_empty(), "{digest}"); + } } #[test] @@ -302,6 +306,38 @@ fn unsupported_engine_is_rejected_before_creating_lease_or_database_files() { assert!(!root.exists()); } +#[test] +fn kernel_with_uncheckpointed_wal_opens_and_preserves_rows() { + let dir = tempfile::tempdir().unwrap(); + drop(KernelStore::open(dir.path()).unwrap()); + + let conn = Connection::open(core_path(dir.path())).unwrap(); + conn.pragma_update(None, "journal_mode", "WAL").unwrap(); + conn.execute( + "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause) + VALUES('t1',1,1,'actor','cause')", + [], + ) + .unwrap(); + let wal = PathBuf::from(format!("{}-wal", core_path(dir.path()).display())); + assert!(wal.is_file(), "the row should still be in the WAL"); + // Leaking skips the clean close that would checkpoint and remove the WAL. + // A crashed writer leaves the uncheckpointed WAL on disk. + std::mem::forget(conn); + + let _store = KernelStore::open(dir.path()).unwrap(); + assert_eq!( + inspect(dir.path(), |conn| conn.query_row( + "SELECT COUNT(*) FROM commit_log", + [], + |row| row.get::<_, i64>(0) + )) + .unwrap(), + 1 + ); + assert!(quarantine_dirs(dir.path()).is_empty()); +} + #[cfg(unix)] fn assert_owner_only(path: &Path, expected: u32) { use std::os::unix::fs::PermissionsExt; diff --git a/crates/mc-store/tests/kernel_schema.rs b/crates/mc-store/tests/kernel_schema.rs index fc515e3b0..1929ea31f 100644 --- a/crates/mc-store/tests/kernel_schema.rs +++ b/crates/mc-store/tests/kernel_schema.rs @@ -1223,15 +1223,21 @@ fn replace_cannot_bypass_the_append_only_guards() { let commit_seq = next_commit(&conn, "tx-replace"); // REPLACE resolves a conflict by deleting the existing row; the delete - // trigger only runs when recursive_triggers is on. - assert!(conn - .execute( - "INSERT OR REPLACE INTO mc_kernel_format_marker( - singleton, format_epoch, database_incarnation_id, schema_digest, created_at - ) VALUES (1, 99, '99999999999999999999999999999999', ?1, 2)", - [PINNED_SCHEMA_DIGEST], - ) - .is_err()); + // trigger only runs when recursive_triggers is on. Every value below + // satisfies the STRICT column types and the length CHECKs, so the BEFORE + // INSERT guard is the only thing left that can reject these statements. + let columns = "singleton, format_epoch, database_incarnation_id, schema_digest, created_at, + marker_digest"; + let values = format!( + "1, 99, '{}', '{}', 2, '{}'", + "9".repeat(32), + PINNED_SCHEMA_DIGEST, + "d".repeat(64) + ); + for verb in ["INSERT OR REPLACE", "REPLACE"] { + let statement = format!("{verb} INTO mc_kernel_format_marker({columns}) VALUES({values})"); + assert!(conn.execute(&statement, []).is_err(), "{statement}"); + } assert_eq!( conn.query_row( "SELECT database_incarnation_id FROM mc_kernel_format_marker", From 5df9599acad00d021f45911b18a63bcaaea5bf0c Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 21:41:00 +0000 Subject: [PATCH 3/6] test(store): reconcile kernel open proofs with the shared application id The kernel now stamps the shared mc application id, so the foreign-family fixture stamped the very id it meant to reject and the header classified it as a kernel database. The fixture uses an id from outside the family and asserts it differs from the kernel's, which fails the test instead of quietly reducing it to a tautology if the constants ever converge again. A database carrying the shared id but a different schema has no header-level tell, so a separate case covers that path and pins the promise that the refusal leaves the family byte-identical. The pinned digest moves because the format marker gained its marker_digest column and its replacement guard. --- crates/mc-store/tests/kernel_open.rs | 29 +++++++++++++++++++++++++- crates/mc-store/tests/kernel_schema.rs | 2 +- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/crates/mc-store/tests/kernel_open.rs b/crates/mc-store/tests/kernel_open.rs index 1a5af53b5..7303121b0 100644 --- a/crates/mc-store/tests/kernel_open.rs +++ b/crates/mc-store/tests/kernel_open.rs @@ -168,10 +168,15 @@ fn every_conclusive_kernel_mismatch_is_quarantined_and_rebuilt() { #[test] fn foreign_family_is_refused_before_sqlite_can_touch_it() { + // `FOREIGN_APPLICATION_ID` must differ from `KERNEL_APPLICATION_ID` so the + // fixture exercises foreign-header classification. + const FOREIGN_APPLICATION_ID: u32 = 0x5A5A_5A5A; + assert_ne!(FOREIGN_APPLICATION_ID, KERNEL_APPLICATION_ID); + let dir = tempfile::tempdir().unwrap(); let path = core_path(dir.path()); let conn = Connection::open(&path).unwrap(); - conn.pragma_update(None, "application_id", 0x4D43_5458_u32) + conn.pragma_update(None, "application_id", FOREIGN_APPLICATION_ID) .unwrap(); conn.execute_batch("CREATE TABLE legacy(value TEXT);") .unwrap(); @@ -192,6 +197,28 @@ fn foreign_family_is_refused_before_sqlite_can_touch_it() { assert!(quarantine_dirs(dir.path()).is_empty()); } +#[test] +fn a_sibling_mc_family_is_refused_and_left_untouched() { + // `KERNEL_APPLICATION_ID` is shared across mc families; schema inspection + // distinguishes them. + let dir = tempfile::tempdir().unwrap(); + let path = core_path(dir.path()); + let conn = Connection::open(&path).unwrap(); + conn.pragma_update(None, "application_id", KERNEL_APPLICATION_ID) + .unwrap(); + conn.execute_batch("CREATE TABLE legacy(value TEXT);") + .unwrap(); + drop(conn); + let before_main = fs::read(&path).unwrap(); + + assert_eq!( + KernelStore::open(dir.path()).unwrap_err(), + KernelError::Inconclusive + ); + assert_eq!(fs::read(&path).unwrap(), before_main); + assert!(quarantine_dirs(dir.path()).is_empty()); +} + #[test] fn malformed_marker_is_inconclusive_and_untouched() { // "g" fails the lowercase-hex check; the all-zero digest fails digest comparison. diff --git a/crates/mc-store/tests/kernel_schema.rs b/crates/mc-store/tests/kernel_schema.rs index 1929ea31f..8a4985fda 100644 --- a/crates/mc-store/tests/kernel_schema.rs +++ b/crates/mc-store/tests/kernel_schema.rs @@ -132,7 +132,7 @@ fn kernel_schema_has_one_ordered_full_shape() { const INCARNATION: &str = "0123456789abcdef0123456789abcdef"; const PINNED_SCHEMA_DIGEST: &str = - "edd21757592088f528212faa058a1c3dd0feb122dc9fa539ecbfa17b67ab6b01"; + "92e7c76e51e721720c2123e0bba45ea41995545b308f9767c64c7aef0fe7a9e6"; #[test] fn kernel_schema_digest_is_pinned_to_the_frozen_v1_shape() { From 98b03f709f0ddf42cf7643f9bd04c2d386603a7f Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 21:53:10 +0000 Subject: [PATCH 4/6] fix(store): run delete triggers on the store's own connections The store built its writer and readers from the pre-classification profile, which set foreign_keys, trusted_schema, and busy_timeout but left recursive_triggers at SQLite's default of off. REPLACE resolves a conflict by deleting the conflicting row, and that delete skips the row's BEFORE DELETE trigger while the pragma is off, so a REPLACE through the writer could rewrite append-only history that the schema declares immutable. The same pragma is already required by verify_kernel_connection_contract, so the store's own connections did not satisfy the contract the crate defines for them. The existing guard test drives a connection from apply_kernel_connection_profile, which sets the pragma, so it could not observe this. The new test asserts the pragma through with_writer and then drives both REPLACE spellings against commit_log. The format marker is read from an untrusted database, where a lookalike table carries no singleton constraint, so the query now bounds both the row count and the identifier and digest widths instead of materializing whatever it finds. --- crates/mc-store/src/kernel/open.rs | 70 +++++++++++++++++++++++++++++- 1 file changed, 69 insertions(+), 1 deletion(-) diff --git a/crates/mc-store/src/kernel/open.rs b/crates/mc-store/src/kernel/open.rs index 9e4681ff0..1e946b6b5 100644 --- a/crates/mc-store/src/kernel/open.rs +++ b/crates/mc-store/src/kernel/open.rs @@ -367,10 +367,16 @@ fn read_valid_marker(conn: &Connection) -> Result { if present.is_none() { return Err(KernelError::Inconclusive); } + // A lookalike table has no singleton constraint, so `LIMIT 2` detects multiple + // rows. let mut statement = conn .prepare( "SELECT format_epoch,database_incarnation_id,schema_digest,created_at,marker_digest - FROM mc_kernel_format_marker", + FROM mc_kernel_format_marker + WHERE length(database_incarnation_id)=32 + AND length(schema_digest)=64 + AND length(marker_digest)=64 + LIMIT 2", ) .map_err(|_| KernelError::Inconclusive)?; let rows = statement @@ -439,6 +445,9 @@ fn apply_preclassification_profile(conn: &Connection) -> rusqlite::Result<()> { conn.pragma_update(None, "foreign_keys", "ON")?; conn.pragma_update(None, "trusted_schema", "OFF")?; conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)?; + // REPLACE deletes the conflicting row, which fires its BEFORE DELETE trigger + // only when recursive_triggers is ON. + conn.pragma_update(None, "recursive_triggers", "ON")?; Ok(()) } @@ -785,4 +794,63 @@ mod tests { }) .unwrap(); } + + #[test] + fn store_connections_run_delete_triggers_for_replace() { + // The schema-level guard test drives a connection built by + // `apply_kernel_connection_profile`, so it cannot observe the pragmas + // `KernelStore` sets on its own writer. + let directory = tempfile::tempdir().unwrap(); + let store = KernelStore::open(directory.path()).unwrap(); + store + .with_writer(|tx| { + assert_eq!( + tx.query_row("PRAGMA recursive_triggers", [], |row| row.get::<_, i64>(0))?, + 1 + ); + tx.execute( + "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause) + VALUES('t1',1,1,'actor','cause')", + [], + )?; + Ok(()) + }) + .unwrap(); + + let commit_seq = store + .with_reader(|tx| { + tx.query_row("SELECT commit_seq FROM commit_log", [], |row| { + row.get::<_, i64>(0) + }) + }) + .unwrap(); + + for verb in ["INSERT OR REPLACE", "REPLACE"] { + let error = store + .with_writer(|tx| { + tx.execute( + &format!( + "{verb} INTO commit_log( + commit_seq,transaction_id,writer_epoch,recorded_at,actor,cause + ) VALUES(?1,'hijack',1,1,'attacker','rewrite')" + ), + [commit_seq], + )?; + Ok(()) + }) + .unwrap_err(); + assert_eq!(error, KernelError::Io, "{verb} must be refused"); + } + + store + .with_reader(|tx| { + assert_eq!( + tx.query_row("SELECT actor FROM commit_log", [], |row| row + .get::<_, String>(0))?, + "actor" + ); + Ok(()) + }) + .unwrap(); + } } From f398525a74a5fdd5368d99e2628180d17ff2dd25 Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 21:54:44 +0000 Subject: [PATCH 5/6] docs(store): state what classification actually preserves The promise said the family is left byte-identical, but a read-only open recreates a missing -shm. The database and its -wal carry the durable content and are preserved; naming that keeps the contract checkable. --- crates/mc-store/src/kernel/open.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/mc-store/src/kernel/open.rs b/crates/mc-store/src/kernel/open.rs index 1e946b6b5..8f9778fcd 100644 --- a/crates/mc-store/src/kernel/open.rs +++ b/crates/mc-store/src/kernel/open.rs @@ -286,10 +286,13 @@ fn expected_identity() -> Result<&'static ExpectedIdentity, KernelError> { EXPECTED_IDENTITY.as_ref().ok_or(KernelError::Io) } -/// Classification must leave the family byte-identical. +/// Classification must leave the database and its `-wal` byte-identical. /// -/// The `Foreign` and `Inconclusive` outcomes promise an untouched family. +/// The `Foreign` and `Inconclusive` outcomes promise untouched durable content. /// Journal recovery and WAL checkpointing would both break that promise. +/// +/// A read-only open may recreate a missing `-shm`; it contains no durable data. +/// SQLite rebuilds the `-shm` from the `-wal` on demand. fn classify_existing_family(path: &Path) -> Result { let expected = expected_identity()?; let mut conn = Connection::open_with_flags( From b34b52e5cbef9378345818db9e37baf526af5909 Mon Sep 17 00:00:00 2001 From: AhravDutta Date: Sun, 30 Aug 2026 22:00:21 +0000 Subject: [PATCH 6/6] test(store): derive the resume fixture from one root spelling The fixture canonicalized db_path but left quarantine_dir as the raw temporary path. resume_quarantine compares marker.db_path to the opened root and valid_quarantine_path compares the two parents, so a root reached through a symlink presented two spellings and the marker was refused as Inconclusive. The store never resumed, and the open failed. Linux /tmp is a real directory, so both spellings matched and the suite passed; macOS reaches its temporary directory through /var -> /private/var, which is where it failed. Reproduced on Linux by pointing TMPDIR at a symlink. --- crates/mc-store/tests/kernel_open.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/mc-store/tests/kernel_open.rs b/crates/mc-store/tests/kernel_open.rs index 7303121b0..782c4540b 100644 --- a/crates/mc-store/tests/kernel_open.rs +++ b/crates/mc-store/tests/kernel_open.rs @@ -249,12 +249,16 @@ fn malformed_marker_is_inconclusive_and_untouched() { #[test] fn valid_interrupted_reset_marker_resumes_without_opening_old_family() { let dir = tempfile::tempdir().unwrap(); - let conn = seed_kernel(dir.path()); + // Quarantine resume compares paths lexically. A symlinked root + // (`/var` -> `/private/var`) fails that comparison when spellings are mixed. + // commentlint: allow(JUDGE) + let root = dir.path().canonicalize().unwrap(); + let conn = seed_kernel(&root); conn.execute_batch("CREATE TABLE unexpected(value INTEGER) STRICT;") .unwrap(); drop(conn); - let db_path = core_path(dir.path()).canonicalize().unwrap(); - let quarantine = dir.path().join("core.sqlite.mc-quarantine-resume"); + let db_path = core_path(&root); + let quarantine = root.join("core.sqlite.mc-quarantine-resume"); fs::create_dir(&quarantine).unwrap(); fs::rename(&db_path, quarantine.join("core.sqlite")).unwrap(); for suffix in ["-journal", "-wal", "-shm"] { @@ -276,13 +280,13 @@ fn valid_interrupted_reset_marker_resumes_without_opening_old_family() { let mut marker = marker_without_digest; marker["marker_digest"] = digest.into(); fs::write( - dir.path().join("core.sqlite.mc-reset"), + root.join("core.sqlite.mc-reset"), serde_json::to_vec(&marker).unwrap(), ) .unwrap(); - let _store = KernelStore::open(dir.path()).unwrap(); - assert!(core_path(dir.path()).is_file()); + let _store = KernelStore::open(&root).unwrap(); + assert!(core_path(&root).is_file()); assert!(quarantine.join("core.sqlite.mc-reset").is_file()); assert_owner_only(&quarantine, 0o700); for name in [