diff --git a/src/db.rs b/src/db.rs index b20399da8..adc52603d 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1532,19 +1532,21 @@ mod tests { let path = dir.path().join("test.graph"); let wal_path = dir.path().join("test.graph.wal"); - let db = Minigraf::open(&path).unwrap(); - db.execute(r#"(transact [[:alice :person/name "Alice"]])"#) - .unwrap(); + { + let db = Minigraf::open(&path).unwrap(); + db.execute(r#"(transact [[:alice :person/name "Alice"]])"#) + .unwrap(); - // WAL should exist before checkpoint - assert!(wal_path.exists(), "WAL must exist after a write"); + // WAL should exist before checkpoint + assert!(wal_path.exists(), "WAL must exist after a write"); - db.checkpoint().unwrap(); + db.checkpoint().unwrap(); - // WAL should be deleted after checkpoint - assert!(!wal_path.exists(), "WAL must be deleted after checkpoint"); + // WAL should be deleted after checkpoint + assert!(!wal_path.exists(), "WAL must be deleted after checkpoint"); + } - // Main file should be present with facts + // Reopen after first handle is dropped (releases file lock) let db2 = Minigraf::open(&path).unwrap(); let facts = db2.inner.fact_storage.get_asserted_facts().unwrap(); assert_eq!(facts.len(), 1, "facts must survive checkpoint"); diff --git a/src/graph/storage.rs b/src/graph/storage.rs index 95f18ef16..d34a7687e 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -450,9 +450,16 @@ pub(crate) fn filter_facts_as_of(facts: Vec, as_of: &AsOf) -> Vec { /// Compute the net-asserted view of a fact set. /// -/// For each unique `(entity, attribute, value)` triple, keeps the fact only if -/// the record with the highest `tx_count` for that triple has `asserted=true`. -/// Facts whose most recent record is a retraction are excluded entirely. +/// For each unique `(entity, attribute, value)` triple: +/// 1. Find the retraction with the highest `tx_count` (if any). +/// 2. Keep all assertions whose `tx_count` is greater than that retraction. +/// 3. Deduplicate surviving assertions by `(valid_from, valid_to)`, keeping the +/// one with the highest `tx_count` for each validity window. +/// +/// This allows the same EAV triple to be asserted at multiple non-overlapping +/// valid-time intervals (e.g., salary=$100k valid 2020–2022 AND 2024–2026). +/// A retraction still cancels all prior assertions of that triple, but +/// re-assertions after the retraction are preserved. /// /// Uses [`encode_value`] for the value key to handle floating-point edge cases /// (NaN canonicalisation, ±0.0 disambiguation) consistently with the rest of @@ -462,25 +469,52 @@ pub(crate) fn filter_facts_as_of(facts: Vec, as_of: &AsOf) -> Vec { /// `get_current_value` and `filter_facts_for_query`. pub(crate) fn net_asserted_facts(facts: Vec) -> Vec { use std::collections::HashMap; - // key: (entity, attribute, canonical value bytes) → fact with highest tx_count - let mut latest: HashMap<(EntityId, Attribute, Vec), Fact> = HashMap::new(); + + // Group all facts by (entity, attribute, canonical value bytes). + let mut groups: HashMap<(EntityId, Attribute, Vec), Vec> = HashMap::new(); for fact in facts { let key = ( fact.entity, fact.attribute.clone(), encode_value(&fact.value), ); - match latest.get(&key) { - None => { - latest.insert(key, fact); + groups.entry(key).or_default().push(fact); + } + + let mut result = Vec::new(); + for (_key, group) in groups { + // Find the highest tx_count among retractions in this group. + let max_retract_tx = group + .iter() + .filter(|f| !f.asserted) + .map(|f| f.tx_count) + .max() + .unwrap_or(0); + + // Keep assertions whose tx_count > max_retract_tx. + // (If no retraction exists, max_retract_tx is 0 and all assertions pass.) + // Deduplicate by (valid_from, valid_to): keep the highest-tx_count assertion + // for each validity window so that re-asserting the same EAV at the same + // window acts as an update rather than a duplicate. + let mut by_window: HashMap<(i64, i64), Fact> = HashMap::new(); + for fact in group { + if !fact.asserted || fact.tx_count <= max_retract_tx { + continue; } - Some(existing) if fact.tx_count > existing.tx_count => { - latest.insert(key, fact); + let window = (fact.valid_from, fact.valid_to); + match by_window.get(&window) { + None => { + by_window.insert(window, fact); + } + Some(existing) if fact.tx_count > existing.tx_count => { + by_window.insert(window, fact); + } + _ => {} } - _ => {} } + result.extend(by_window.into_values()); } - latest.into_values().filter(|f| f.asserted).collect() + result } /// Resolve a FactRef to a Fact using the committed reader (for on-disk facts) diff --git a/src/query/datalog/parser.rs b/src/query/datalog/parser.rs index 9baba4c9d..2b750e9b6 100644 --- a/src/query/datalog/parser.rs +++ b/src/query/datalog/parser.rs @@ -148,9 +148,15 @@ fn tokenize(input: &str) -> Result, String> { let mut num_str = String::from("-"); let (is_float, num) = parse_number(&mut chars, &mut num_str)?; if is_float { - tokens.push(Token::Float(num.parse().unwrap())); + let f: f64 = num + .parse() + .map_err(|_| format!("Float literal out of range: {}", num))?; + tokens.push(Token::Float(f)); } else { - tokens.push(Token::Integer(num.parse().unwrap())); + let i: i64 = num + .parse() + .map_err(|_| format!("Integer literal out of range: {}", num))?; + tokens.push(Token::Integer(i)); } } else { // It's a symbol starting with - @@ -169,9 +175,15 @@ fn tokenize(input: &str) -> Result, String> { let mut num_str = String::new(); let (is_float, num) = parse_number(&mut chars, &mut num_str)?; if is_float { - tokens.push(Token::Float(num.parse().unwrap())); + let f: f64 = num + .parse() + .map_err(|_| format!("Float literal out of range: {}", num))?; + tokens.push(Token::Float(f)); } else { - tokens.push(Token::Integer(num.parse().unwrap())); + let i: i64 = num + .parse() + .map_err(|_| format!("Integer literal out of range: {}", num))?; + tokens.push(Token::Integer(i)); } } // Symbols (including variables starting with ?) @@ -2810,6 +2822,43 @@ mod tests { let result = parse_edn("$"); assert!(result.is_err(), "bare '$' should be a parse error"); } + + #[test] + fn test_integer_overflow_returns_error() { + let result = tokenize("99999999999999999999999999999"); + assert!(result.is_err(), "i64 overflow must return error, not panic"); + let err = result.unwrap_err(); + assert!( + err.contains("out of range"), + "error should mention out of range" + ); + } + + #[test] + fn test_negative_integer_overflow_returns_error() { + let result = tokenize("-99999999999999999999999999999"); + assert!(result.is_err(), "negative i64 overflow must return error"); + } + + #[test] + fn test_float_overflow_returns_error() { + // f64::MAX is ~1.8e308, so 1e999 overflows to infinity + let result = tokenize("1.0e999"); + // f64 parse of very large exponent may produce infinity rather than error; + // our code rejects it if parse() itself errors. For f64, parse() succeeds + // with infinity, so this may not error. The critical fix is for integers. + // Just ensure no panic occurs. + let _ = result; + } + + #[test] + fn test_overflow_in_transact_returns_error() { + let result = parse_datalog_command("(transact [[:e :a 99999999999999999999999999999]])"); + assert!( + result.is_err(), + "overflow in transact should be parse error" + ); + } } #[cfg(test)] diff --git a/src/storage/backend/file.rs b/src/storage/backend/file.rs index c94404686..c238959c3 100644 --- a/src/storage/backend/file.rs +++ b/src/storage/backend/file.rs @@ -5,6 +5,86 @@ use std::fs::{File, OpenOptions}; use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; +/// Advisory file lock to prevent multi-process corruption. +/// +/// Uses a sidecar `.lock` file with exclusive creation semantics. +/// The lock is released (file deleted) on drop. +struct FileLock { + path: PathBuf, +} + +impl FileLock { + /// Attempt to acquire an exclusive lock for the given database path. + /// Returns `Err` if another process already holds the lock. + /// + /// If a stale lock file exists (the holder PID is no longer running), + /// it is automatically removed and a new lock is acquired. This handles + /// the case where the previous process crashed without cleaning up. + fn acquire(db_path: &Path) -> Result { + let lock_path = db_path.with_extension("graph.lock"); + // create_new fails with AlreadyExists if the lock file is present + match OpenOptions::new() + .write(true) + .create_new(true) + .open(&lock_path) + { + Ok(mut f) => { + // Write PID for diagnostics (best-effort) + let _ = write!(f, "{}", std::process::id()); + Ok(FileLock { path: lock_path }) + } + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { + // Check if the holder process is still alive + let holder = std::fs::read_to_string(&lock_path).unwrap_or_default(); + if let Ok(pid) = holder.trim().parse::() { + let our_pid = std::process::id(); + if pid == our_pid || !Self::is_process_alive(pid) { + // Stale lock — either our own leaked handle or previous + // process crashed. Remove and retry. + let _ = std::fs::remove_file(&lock_path); + return Self::acquire(db_path); + } + } + anyhow::bail!( + "Database is locked by another process (lock file: {}, holder PID: {}). \ + If no other process is using this database, delete the lock file manually.", + lock_path.display(), + holder.trim() + ); + } + Err(e) => { + anyhow::bail!( + "Failed to acquire database lock at {}: {}", + lock_path.display(), + e + ); + } + } + } + + /// Check if a process with the given PID is still running. + fn is_process_alive(pid: u32) -> bool { + // On Linux/Android, /proc/ exists iff the process is alive. + let proc_path = format!("/proc/{}", pid); + if std::path::Path::new(&proc_path).exists() { + return true; + } + // On Linux, if /proc exists but /proc/ doesn't, the process is dead. + if std::path::Path::new("/proc").exists() { + return false; + } + // On non-procfs systems (macOS, Windows), assume alive (conservative). + // Users must manually delete stale lock files on these platforms. + true + } +} + +impl Drop for FileLock { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + /// File-based storage backend for native platforms. /// /// Stores graph data in a single `.graph` file with a page-based structure: @@ -22,6 +102,7 @@ pub struct FileBackend { file: File, header: FileHeader, is_new: bool, + _lock: FileLock, } impl FileBackend { @@ -29,9 +110,16 @@ impl FileBackend { /// /// If the file doesn't exist, creates it with an initial header. /// If it exists, validates and loads the header. + /// + /// Acquires an advisory file lock (sidecar `.graph.lock` file) to prevent + /// multi-process corruption. Returns an error if the database is already + /// opened by another process. pub fn open>(path: P) -> Result { let path = path.as_ref().to_path_buf(); + // Acquire advisory lock before touching the database file. + let lock = FileLock::acquire(&path)?; + let mut file = OpenOptions::new() .read(true) .write(true) @@ -71,6 +159,7 @@ impl FileBackend { file, header, is_new, + _lock: lock, }) } diff --git a/src/storage/persistent_facts.rs b/src/storage/persistent_facts.rs index 4a57dce4c..a149f193c 100644 --- a/src/storage/persistent_facts.rs +++ b/src/storage/persistent_facts.rs @@ -239,14 +239,30 @@ impl PersistentFactStorage { self.committed_fact_pages .store(num_fact_pages, Ordering::SeqCst); - // Compute page-based checksum to verify indexes are valid - let computed = { + // Compute page-based checksum to verify data integrity. + // New files (post-fix): checksum covers ALL pages (facts + indexes). + // Old files (pre-fix): checksum covers only fact pages. + // Try full checksum first; fall back to fact-only for backwards compat. + let needs_rebuild = if num_fact_pages == 0 || header.eavt_root_page == 0 { + num_fact_pages > 0 // rebuild if facts exist but no index root + } else { let backend = self.backend.lock().unwrap(); - compute_page_checksum(&*backend, 1, num_fact_pages)? + let stored = header.index_checksum; + // Total data pages: pages 1 through page_count-1 (everything except header) + let total_data_pages = header.page_count.saturating_sub(1); + let full_checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + if full_checksum == stored { + false // new-style checksum matches: facts + indexes verified + } else { + // Fall back: old files stored checksum over fact pages only + let fact_checksum = compute_page_checksum(&*backend, 1, num_fact_pages)?; + if fact_checksum == stored { + false // old-style checksum matches: facts verified, indexes unprotected + } else { + true // neither matches: corruption detected, rebuild + } + } }; - let stored = header.index_checksum; - let needs_rebuild = - num_fact_pages > 0 && (computed != stored || header.eavt_root_page == 0); // Register CommittedFactReader on FactStorage (before WAL replay) let loader: std::sync::Arc = @@ -307,7 +323,10 @@ impl PersistentFactStorage { next3, )?; - // Write v6 header + // Write header with full-coverage checksum (facts + indexes) + let total_data_pages = next4.saturating_sub(1); + let full_checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + let mut new_header = FileHeader::new(); new_header.page_count = next4; new_header.node_count = all_facts.len() as u64; @@ -316,7 +335,7 @@ impl PersistentFactStorage { new_header.aevt_root_page = aevt_root; new_header.avet_root_page = avet_root; new_header.vaet_root_page = vaet_root; - new_header.index_checksum = computed; + new_header.index_checksum = full_checksum; new_header.fact_page_format = FACT_PAGE_FORMAT_PACKED; new_header.fact_page_count = num_fact_pages; @@ -533,11 +552,17 @@ impl PersistentFactStorage { .store(validated_num_fact_pages, Ordering::SeqCst); // Verify index integrity via checksum before trusting the old indexes. - // If checksum doesn't match, rebuild indexes from facts instead. + // Try full checksum (facts + indexes) first, then fact-only for old files. let use_old_indexes = validated_num_fact_pages > 0 && header.index_checksum > 0 && { let backend = self.backend.lock().unwrap(); - let computed = compute_page_checksum(&*backend, 1, validated_num_fact_pages)?; - computed == header.index_checksum + let total_data_pages = header.page_count.saturating_sub(1); + let full = compute_page_checksum(&*backend, 1, total_data_pages)?; + if full == header.index_checksum { + true + } else { + let fact_only = compute_page_checksum(&*backend, 1, validated_num_fact_pages)?; + fact_only == header.index_checksum + } }; let (eavt, aevt, avet, vaet) = if use_old_indexes { @@ -668,8 +693,9 @@ impl PersistentFactStorage { new_header.aevt_root_page = aevt_root; new_header.avet_root_page = avet_root; new_header.vaet_root_page = vaet_root; - // Recompute the checksum for the new indexes - let computed_checksum = compute_page_checksum(&*backend, 1, validated_num_fact_pages)?; + // Checksum over all data pages (facts + indexes) + let total_data_pages = final_next_free.saturating_sub(1); + let computed_checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; new_header.index_checksum = computed_checksum; new_header.fact_page_format = header.fact_page_format; new_header.fact_page_count = validated_num_fact_pages; @@ -784,8 +810,10 @@ impl PersistentFactStorage { } let new_total_fact_pages = old_fact_page_count + new_pages.len() as u64; - // CRC32 over ALL fact pages (old committed + newly appended) - let checksum = compute_page_checksum(&*backend, 1, new_total_fact_pages)?; + // Sync fact pages to disk before building indexes on top of them. + // Without this, a crash during index build could leave partially-flushed + // fact pages that the old header's index roots would try to traverse. + backend.sync()?; // ── Step C: build sorted index entries for pending facts ──────────────── let (pending_eavt, pending_aevt, pending_avet, pending_vaet) = @@ -830,7 +858,18 @@ impl PersistentFactStorage { let (vaet_root, next4) = build_btree(vaet_ser.into_iter(), &mut *backend, &self.page_cache, next3)?; - // ── Step E: write v6 header (last write = crash-safe boundary) ────────── + // Sync index pages to disk before writing the header. + // The header update is the atomic commit point: once it's durable, + // recovery uses the new root pages. All data those roots reference + // must already be on stable storage. + backend.sync()?; + + // CRC32 over ALL data pages (facts + indexes), excluding page 0 (header). + // This detects corruption in both fact pages and B+tree index pages. + let total_data_pages = next4.saturating_sub(1); + let checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + + // ── Step E: write header (last write = crash-safe boundary) ───────────── let mut header = FileHeader::new(); // version=7 header.page_count = next4; header.node_count = curr_header.node_count + pending_facts.len() as u64; diff --git a/tests/bitemporal_test.rs b/tests/bitemporal_test.rs index 932715230..2a206bf13 100644 --- a/tests/bitemporal_test.rs +++ b/tests/bitemporal_test.rs @@ -359,3 +359,99 @@ fn test_as_of_counter_time_travel() { ); assert_eq!(result_rows(result3).len(), 3, "as-of 3: name + age + city"); } + +// ============================================================================ +// Regression: same EAV asserted at multiple valid-time intervals must coexist +// ============================================================================ + +/// The same (entity, attribute, value) triple asserted with different valid-time +/// windows must all be visible when querying within their respective windows. +/// Previously, `net_asserted_facts` collapsed them by keeping only the latest +/// tx_count, causing earlier valid-time intervals to be silently lost. +#[test] +fn test_same_eav_multiple_valid_time_intervals() { + let db = Minigraf::in_memory().unwrap(); + + // tx_count=1: Alice earns 100000 from 2020-01-01 to 2022-01-01 + exec( + &db, + r#"(transact {:valid-from "2020-01-01" :valid-to "2022-01-01"} [[:alice :salary 100000]])"#, + ); + + // tx_count=2: Alice earns 100000 again from 2024-01-01 to 2026-01-01 + exec( + &db, + r#"(transact {:valid-from "2024-01-01" :valid-to "2026-01-01"} [[:alice :salary 100000]])"#, + ); + + // Query at 2021-06-01 → should find the 2020-2022 assertion + let rows_2021 = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at "2021-06-01" :where [:alice :salary ?v]])"#, + )); + assert_eq!(rows_2021.len(), 1, "salary visible at 2021-06-01"); + assert_eq!(rows_2021[0][0], Value::Integer(100000)); + + // Query at 2025-01-01 → should find the 2024-2026 assertion + let rows_2025 = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at "2025-01-01" :where [:alice :salary ?v]])"#, + )); + assert_eq!(rows_2025.len(), 1, "salary visible at 2025-01-01"); + assert_eq!(rows_2025[0][0], Value::Integer(100000)); + + // Query at 2023-01-01 → gap between intervals, should find nothing + let rows_2023 = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at "2023-01-01" :where [:alice :salary ?v]])"#, + )); + assert_eq!(rows_2023.len(), 0, "no salary in the gap between intervals"); + + // Query with :any-valid-time → should see BOTH intervals + let rows_any = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at :any-valid-time :where [:alice :salary ?v]])"#, + )); + assert_eq!( + rows_any.len(), + 2, + "both valid-time intervals visible with :any-valid-time" + ); +} + +/// Retraction still cancels all prior assertions of the same EAV, but +/// a re-assertion after the retraction is preserved. +#[test] +fn test_retraction_cancels_prior_intervals_reassertion_survives() { + let db = Minigraf::in_memory().unwrap(); + + // tx_count=1: salary valid 2020-2022 + exec( + &db, + r#"(transact {:valid-from "2020-01-01" :valid-to "2022-01-01"} [[:alice :salary 100000]])"#, + ); + + // tx_count=2: retract the salary (cancels all prior assertions of this EAV) + exec(&db, r#"(retract [[:alice :salary 100000]])"#); + + // tx_count=3: re-assert salary for 2024-2026 + exec( + &db, + r#"(transact {:valid-from "2024-01-01" :valid-to "2026-01-01"} [[:alice :salary 100000]])"#, + ); + + // The 2020-2022 assertion should be gone (retracted at tx_count=2) + let rows_2021 = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at "2021-06-01" :where [:alice :salary ?v]])"#, + )); + assert_eq!(rows_2021.len(), 0, "retracted interval no longer visible"); + + // The 2024-2026 re-assertion should survive (tx_count=3 > retraction tx_count=2) + let rows_2025 = result_rows(exec( + &db, + r#"(query [:find ?v :valid-at "2025-01-01" :where [:alice :salary ?v]])"#, + )); + assert_eq!(rows_2025.len(), 1, "re-assertion after retraction visible"); + assert_eq!(rows_2025[0][0], Value::Integer(100000)); +} diff --git a/tests/btree_v6_test.rs b/tests/btree_v6_test.rs index 2ed6427e5..b2a813132 100644 --- a/tests/btree_v6_test.rs +++ b/tests/btree_v6_test.rs @@ -203,10 +203,12 @@ fn test_v6_dead_pages_queries_correct_after_two_checkpoints() { populate_and_checkpoint(20, &path); // Second checkpoint (adds new facts + new B+tree; old index pages are dead) - let db = OpenOptions::new().path(&path).open().unwrap(); - db.execute("(transact [[:e20 :val 20][:e21 :val 21]])") - .unwrap(); - db.checkpoint().unwrap(); + { + let db = OpenOptions::new().path(&path).open().unwrap(); + db.execute("(transact [[:e20 :val 20][:e21 :val 21]])") + .unwrap(); + db.checkpoint().unwrap(); + } // Re-open and verify queries are correct let db2 = OpenOptions::new().path(&path).open().unwrap();