diff --git a/CHANGELOG.md b/CHANGELOG.md index ac7d40c71..e6458fb63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- **A query on another thread sees each transaction whole (#554).** A commit wrote its WAL entry, then added the transaction's facts to the in-memory store one fact at a time, each under its own lock, so a query running on another thread could return part of the transaction. `execute("(transact ...)")`, `execute("(retract ...)")` and `WriteTransaction::commit` all did this. They now add the whole transaction under one lock, as does WAL replay. Stored data was never affected. Affected `WriteTransaction` since v0.5.0 and `execute` since v0.6.0. Also fixed in v3.0.0. +- **All patterns of one query read the same state of the database (#555).** A query whose patterns name entities or attributes read each of them with its own lookup, each under its own lock, so a transaction that committed between two lookups was seen by some patterns and not others, and the query could return rows that match no state of the database. The lookups now run under one lock. Stored data was never affected. Affected every release since v1.1.0. Also fixed in v3.0.0. +- **A query during a checkpoint no longer misses committed facts (#558).** A checkpoint wrote its new fact pages over the index pages that queries on other threads were still reading, then installed its new readers and cleared the pending facts in three separately locked steps. A query during a checkpoint, including the automatic one every `wal_checkpoint_threshold` WAL entries, could miss facts committed before it started or see part of a transaction. Queries now wait for a checkpoint to finish: it holds the store's write lock from its first page write until the new state is published. Stored data was never affected. Affected every release since v0.9.0; v3.0.0's copy-on-write checkpoints are not affected. + +### Tests + +- `tests/snapshot_atomicity_test.rs` races a writer against a reader: a query must see each transaction whole, whether written with `execute` or a `WriteTransaction` (#554), and must read one state for all its patterns (#555), in memory and on file, and must not miss committed facts while a checkpoint runs (#558). + ## v2.0.5 — 2026-10-11 Patch release on the v2.x line under the [support policy](PHILOSOPHY.md#support-policy): two data-integrity fixes (#506, #513) and three bug fixes backported from v3.0.0 (#528, #457, #476), plus a weekly soak test and working coverage gates. File format is unchanged (v7). Two new error codes, `STG-045` and `WAL-007`, numbered as in v3.0.0; no other API changes. Upgrading is recommended for every v2.x user. diff --git a/CLAUDE.md b/CLAUDE.md index 66de66d99..d14476e07 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -174,7 +174,7 @@ Auto-migrates v1/v2/v3/v4/v5/v6 → v7 on open/checkpoint. ## Test Coverage -**1241 tests** (1232 passing, 9 ignored; unit + integration + doc). +**1250 tests** (1241 passing, 9 ignored; unit + integration + doc). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/README.md b/README.md index a952b4310..9232f34a5 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit ```bash cargo run # interactive Datalog REPL -cargo test # run 1241 tests +cargo test # run 1250 tests cargo run < demos/demo_recursive.txt # recursive rules demo ``` diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 89ebe1621..8ef6a1a4c 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,7 +4,7 @@ **Verified**: 2026-10-11 with `cargo test` -**Result**: 1,232 passing tests and 9 ignored tests (1,241 total), including the golden-file corpus (#391) +**Result**: 1,241 passing tests and 9 ignored tests (1,250 total), including the golden-file corpus (#391) For v2.0.5, `tests/edge_cases_test.rs` checks that a file of 1 to 4,095 bytes is `STG-045`, unchanged and with no WAL, when it is a database cut below one page or foreign bytes, and opens as a new database that takes a write and reads it back after reopen when it is a prefix of the initial header or all zeros (#506). @@ -22,7 +22,7 @@ The v2.0.5 suite covers the public database API, the Datalog engine, storage and - **Core database behavior**: in-memory and file-backed operation, transactions, checkpoints, retractions, bi-temporal queries, recursive rules, prepared statements, aggregation, expressions, disjunction, window functions, UDFs, and magic-sets evaluation. - **Storage correctness**: packed pages, B+tree indexes and range scans, cache behavior, file-header validation, migration, WAL replay, checksums, corruption handling, and fault injection. -- **Reliability and concurrency**: concurrent reads and writes, rollback behavior, same-process handle exclusion, cross-process locking, PID namespaces, NFS and container lock behavior, and SIGKILL recovery. +- **Reliability and concurrency**: concurrent reads and writes, rollback behavior, same-process handle exclusion, cross-process locking, PID namespaces, NFS and container lock behavior, and SIGKILL recovery, and queries seeing each concurrent transaction whole and reading one state for all their patterns, also during a checkpoint (`tests/snapshot_atomicity_test.rs`, #554, #555, #558). - **Public diagnostics**: structured parser, query, storage, WAL, API, and internal error codes; the error-code registry is checked against [`ERROR_REFERENCE.md`](ERROR_REFERENCE.md). - **Compatibility and quality**: golden-file corpus of released file formats, cross-platform format compatibility, property-based tests, long-haul smoke coverage, a weekly soak at 10M facts with a 5-hour churn (#392), XTDB/Datomic semantic compatibility, grammar conformance, and rustdoc examples. diff --git a/src/browser/mod.rs b/src/browser/mod.rs index 66f27b32f..a7d80ca3b 100644 --- a/src/browser/mod.rs +++ b/src/browser/mod.rs @@ -267,12 +267,11 @@ impl BrowserDb { }) .collect(); - for fact in &stamped { - inner - .fact_storage - .load_fact(fact.clone()) - .map_err(|e| JsValue::from_str(&e.to_string()))?; - } + // One batch, so no query sees part of the transaction (#554). + inner + .fact_storage + .load_facts(stamped.iter().cloned()) + .map_err(|e| JsValue::from_str(&e.to_string()))?; inner.pfs.mark_dirty(); inner diff --git a/src/db.rs b/src/db.rs index b2d43d2cf..66059e09e 100644 --- a/src/db.rs +++ b/src/db.rs @@ -519,9 +519,7 @@ impl Minigraf { // Already present in the main file; skip. continue; } - for fact in &entry.facts { - let _ = fact_storage.load_fact(fact.clone())?; - } + let _ = fact_storage.load_facts(entry.facts.iter().cloned())?; #[allow(clippy::arithmetic_side_effects)] { replayed += 1; @@ -637,10 +635,12 @@ impl Minigraf { &stamped, )?; - // WAL succeeded — now apply facts to shared FactStorage. - for fact in &stamped { - let _ = self.inner.fact_storage.load_fact(fact.clone())?; - } + // WAL succeeded — now apply facts to shared FactStorage, in one + // batch so no query sees part of the transaction (#554). + let _ = self + .inner + .fact_storage + .load_facts(stamped.iter().cloned())?; // Trigger auto-checkpoint AFTER facts are in FactStorage so the // checkpoint captures the newly written facts. @@ -1249,10 +1249,9 @@ impl<'a> WriteTransaction<'a> { &stamped, )?; - // WAL succeeded — now apply facts to shared FactStorage. - for fact in stamped { - let _ = self.inner.fact_storage.load_fact(fact)?; - } + // WAL succeeded — now apply facts to shared FactStorage, in one + // batch so no query sees part of the transaction (#554). + let _ = self.inner.fact_storage.load_facts(stamped)?; // Trigger auto-checkpoint AFTER facts are in FactStorage so the // checkpoint captures the newly written facts. diff --git a/src/graph/storage.rs b/src/graph/storage.rs index 1edaa6354..90aa0aba9 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -294,6 +294,34 @@ impl FactStorage { Ok(true) } + /// Insert facts with their tx_id and tx_count preserved, all under one + /// write lock, so a query sees either none of them or all of them (#554). + /// + /// Commits and WAL replay use this for a whole transaction. Duplicates are + /// skipped as in [`Self::load_fact`]. Returns how many facts were loaded. + pub(crate) fn load_facts(&self, facts: impl IntoIterator) -> Result { + let mut d = self + .data + .write() + .map_err(|_| err_coded!(ErrorCode::Int050, "data"))?; + // Pick the new facts and their slots before changing anything, so a + // slot past `u32::MAX` fails the whole batch, never part of it. + let mut keys = HashSet::new(); + let new: Vec<(PendingKey, Fact)> = facts + .into_iter() + .map(|f| (pending_key(&f), f)) + .filter(|(k, _)| !d.pending_keys.contains(k) && keys.insert(k.clone())) + .collect(); + let refs = pending_refs(d.facts.len(), new.len())?; + let loaded = new.len(); + for ((key, fact), fact_ref) in new.into_iter().zip(refs) { + d.pending_keys.insert(key); + d.pending_indexes.insert(&fact, fact_ref); + d.facts.push(fact); + } + Ok(loaded) + } + /// Raise tx_counter to max(tx_count) across all loaded facts. /// /// Must be called after all `load_fact()` calls complete so that the next @@ -337,14 +365,7 @@ impl FactStorage { .data .read() .map_err(|_| err_coded!(ErrorCode::Int050, "data"))?; - let mut all = Vec::new(); - // Committed facts first (on disk, via CommittedFactReader) - if let Some(loader) = &d.committed { - all.extend(loader.stream_all()?); - } - // Then pending facts (post-checkpoint, in memory) - all.extend(d.facts.iter().cloned()); - Ok(all) + all_facts(&d) } /// Return all facts visible as of the given transaction point. @@ -390,13 +411,22 @@ impl FactStorage { d.pending_indexes = indexes; } + /// Exclusive access for a checkpoint (#558): no query reads the store + /// until the guard drops, so none sees the pages the checkpoint overwrites + /// or a state between its publishing steps. + pub(crate) fn lock_for_checkpoint(&self) -> CheckpointGuard<'_> { + CheckpointGuard(self.data.write().unwrap_or_else(|e| e.into_inner())) + } + /// Return the pending (uncommitted) facts held in memory. + #[cfg(test)] pub(crate) fn get_pending_facts(&self) -> Vec { let d = self.data.read().unwrap_or_else(|e| e.into_inner()); d.facts.clone() } /// Clear pending facts and pending indexes after a successful checkpoint. + #[cfg(test)] pub(crate) fn post_checkpoint_clear(&self) { let mut d = self.data.write().unwrap_or_else(|e| e.into_inner()); d.facts.clear(); @@ -596,74 +626,13 @@ fn resolve_fact_ref(d: &FactData, fr: FactRef) -> Result { /// Production helpers on FactStorage: index-driven entity/attribute lookups used by the query executor. impl FactStorage { + #[cfg(test)] /// Get all facts for a specific entity (index-driven). pub(crate) fn get_facts_by_entity(&self, entity_id: &EntityId) -> Result> { - use crate::storage::index::EavtKey; - let d = self.data.read().unwrap_or_else(|e| e.into_inner()); - - let start = EavtKey { - entity: *entity_id, - attribute: String::new(), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; - let next_entity = uuid::Uuid::from_u128(entity_id.as_u128().wrapping_add(1)); - let end = EavtKey { - entity: next_entity, - attribute: String::new(), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; - - // Fallback: no indexes built yet - if d.pending_indexes.eavt.is_empty() && d.committed_index_reader.is_none() { - if d.committed.is_none() { - return Ok(d - .facts - .iter() - .filter(|f| &f.entity == entity_id) - .cloned() - .collect()); - } - let mut result: Vec = d - .facts - .iter() - .filter(|f| &f.entity == entity_id) - .cloned() - .collect(); - if let Some(loader) = &d.committed { - for fact in loader.stream_all()? { - if &fact.entity == entity_id { - result.push(fact); - } - } - } - return Ok(result); - } - - let mut facts = Vec::new(); - - // Pending: in-memory BTreeMap bounded range. - for (key, &fr) in d.pending_indexes.eavt.range(start.clone()..end.clone()) { - if key.entity != *entity_id { - break; - } - facts.push(resolve_fact_ref(&d, fr)?); - } - - // Committed: on-disk B+tree range scan - if let Some(reader) = &d.committed_index_reader { - let committed_refs = reader.range_scan_eavt(&start, Some(&end))?; - for fr in committed_refs { - facts.push(resolve_fact_ref(&d, fr)?); - } - } - - Ok(facts) + self.get_facts_for_lookups(&[Lookup::Entity(entity_id)]) } + #[cfg(test)] /// Get every stored record for one `(entity, attribute)` pair (index-driven, #323). /// /// Range-scans EAVT over exactly `[(e, a, …), (e, a + "\0", …))`, so other @@ -674,116 +643,264 @@ impl FactStorage { entity_id: &EntityId, attribute: &Attribute, ) -> Result> { - use crate::storage::index::EavtKey; - let d = self.data.read().unwrap_or_else(|e| e.into_inner()); - let matches = |f: &Fact| &f.entity == entity_id && &f.attribute == attribute; + self.get_facts_for_lookups(&[Lookup::EntityAttribute(entity_id, attribute)]) + } - // Fallback: no indexes built yet - if d.pending_indexes.eavt.is_empty() && d.committed_index_reader.is_none() { - let mut result: Vec = d.facts.iter().filter(|f| matches(f)).cloned().collect(); - if let Some(loader) = &d.committed { - result.extend(loader.stream_all()?.into_iter().filter(|f| matches(f))); - } - return Ok(result); + #[cfg(test)] + /// Get all facts for a specific attribute (index-driven). + pub(crate) fn get_facts_by_attribute(&self, attribute: &Attribute) -> Result> { + self.get_facts_for_lookups(&[Lookup::Attribute(attribute)]) + } + + /// The facts of several lookups, all read from one state of the store: no + /// commit lands between two of them (#555). Results are concatenated in + /// `lookups` order and not deduplicated. + pub(crate) fn get_facts_for_lookups(&self, lookups: &[Lookup<'_>]) -> Result> { + let d = self.data.read().unwrap_or_else(|e| e.into_inner()); + let mut facts = Vec::new(); + for lookup in lookups { + facts.extend(match *lookup { + Lookup::Entity(e) => entity_facts(&d, e)?, + Lookup::EntityAttribute(e, a) => entity_attribute_facts(&d, e, a)?, + Lookup::Attribute(a) => attribute_facts(&d, a)?, + }); } + Ok(facts) + } +} - let start = EavtKey { - entity: *entity_id, - attribute: attribute.clone(), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; - // Exact successor of `attribute`: the smallest string that sorts after it is - // `attribute + "\0"`, so `[start, end)` holds exactly this (e, a) pair. Always - // valid UTF-8 — unlike incrementing the last byte, which fails for attributes - // ending in 0x7F or a character whose last UTF-8 byte is 0xBF (e.g. `:丿`). - let end = EavtKey { - entity: *entity_id, - attribute: format!("{attribute}\0"), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; +/// [`FactStorage`] held exclusively by a checkpoint; see +/// [`FactStorage::lock_for_checkpoint`]. Take it before the backend lock: a +/// query holds the store's read lock while it reads pages. +pub(crate) struct CheckpointGuard<'a>(std::sync::RwLockWriteGuard<'a, FactData>); - let mut facts = Vec::new(); +impl CheckpointGuard<'_> { + /// The pending (uncommitted) facts held in memory. + pub(crate) fn pending_facts(&self) -> Vec { + self.0.facts.clone() + } - // Pending: EAVT keys for the exact (e, a) are contiguous from `start`; - // the first key with a different entity or attribute ends the run. - for (key, &fr) in d.pending_indexes.eavt.range(start.clone()..) { - if key.entity != *entity_id || key.attribute != *attribute { - break; - } - facts.push(resolve_fact_ref(&d, fr)?); - } + /// Install the checkpoint's readers and drop the pending facts it wrote, + /// in one step. + pub(crate) fn publish( + &mut self, + reader: Arc, + index_reader: Arc, + ) { + let d = &mut *self.0; + d.committed = Some(reader); + d.committed_index_reader = Some(index_reader); + d.facts.clear(); + d.pending_keys.clear(); + d.pending_indexes = Indexes::new(); + } +} - // Committed: on-disk B+tree range scan over exactly this (e, a) pair; the - // post-filter is a cheap guard, not something the range relies on. - if let Some(reader) = &d.committed_index_reader { - for fr in reader.range_scan_eavt(&start, Some(&end))? { - let fact = resolve_fact_ref(&d, fr)?; - if matches(&fact) { - facts.push(fact); +/// One index lookup of [`FactStorage::get_facts_for_lookups`]. +#[derive(Clone, Copy)] +pub(crate) enum Lookup<'a> { + /// Every record of one entity. + Entity(&'a EntityId), + /// Every record of one `(entity, attribute)` pair. + EntityAttribute(&'a EntityId, &'a Attribute), + /// Every record with one attribute. + Attribute(&'a Attribute), +} + +/// Committed facts, then pending facts. The caller holds `d`'s lock. +fn all_facts(d: &FactData) -> Result> { + let mut all = Vec::new(); + // Committed facts first (on disk, via CommittedFactReader) + if let Some(loader) = &d.committed { + all.extend(loader.stream_all()?); + } + // Then pending facts (post-checkpoint, in memory) + all.extend(d.facts.iter().cloned()); + Ok(all) +} + +/// Get all facts for a specific entity (index-driven). +fn entity_facts(d: &FactData, entity_id: &EntityId) -> Result> { + use crate::storage::index::EavtKey; + + let start = EavtKey { + entity: *entity_id, + attribute: String::new(), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + let next_entity = uuid::Uuid::from_u128(entity_id.as_u128().wrapping_add(1)); + let end = EavtKey { + entity: next_entity, + attribute: String::new(), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + + // Fallback: no indexes built yet + if d.pending_indexes.eavt.is_empty() && d.committed_index_reader.is_none() { + if d.committed.is_none() { + return Ok(d + .facts + .iter() + .filter(|f| &f.entity == entity_id) + .cloned() + .collect()); + } + let mut result: Vec = d + .facts + .iter() + .filter(|f| &f.entity == entity_id) + .cloned() + .collect(); + if let Some(loader) = &d.committed { + for fact in loader.stream_all()? { + if &fact.entity == entity_id { + result.push(fact); } } } - - Ok(facts) + return Ok(result); } - /// Get all facts for a specific attribute (index-driven). - pub(crate) fn get_facts_by_attribute(&self, attribute: &Attribute) -> Result> { - use crate::storage::index::AevtKey; - let d = self.data.read().unwrap_or_else(|e| e.into_inner()); + let mut facts = Vec::new(); - // Fallback: no index - if d.pending_indexes.aevt.is_empty() && d.committed_index_reader.is_none() { - drop(d); - return Ok(self - .get_all_facts()? - .into_iter() - .filter(|f| &f.attribute == attribute) - .collect()); + // Pending: in-memory BTreeMap bounded range. + for (key, &fr) in d.pending_indexes.eavt.range(start.clone()..end.clone()) { + if key.entity != *entity_id { + break; } + facts.push(resolve_fact_ref(d, fr)?); + } - let start = AevtKey { - attribute: attribute.clone(), - entity: uuid::Uuid::nil(), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; - // Exact successor of `attribute` (see `get_facts_by_entity_attribute_indexed`): - // `[start, end)` holds exactly this attribute, never prefix siblings such as - // `:ab` for `:a`, and is valid UTF-8 for any attribute (#381). - let end = AevtKey { - attribute: format!("{attribute}\0"), - entity: uuid::Uuid::nil(), - valid_from: i64::MIN, - valid_to: i64::MIN, - tx_count: 0, - }; + // Committed: on-disk B+tree range scan + if let Some(reader) = &d.committed_index_reader { + let committed_refs = reader.range_scan_eavt(&start, Some(&end))?; + for fr in committed_refs { + facts.push(resolve_fact_ref(d, fr)?); + } + } - let mut facts = Vec::new(); + Ok(facts) +} - // Pending: AEVT keys for the exact attribute are contiguous from `start`. - for (_, &fr) in d.pending_indexes.aevt.range(start.clone()..end.clone()) { - facts.push(resolve_fact_ref(&d, fr)?); +/// Get every stored record for one `(entity, attribute)` pair (index-driven, #323). +/// +/// Range-scans EAVT over exactly `[(e, a, …), (e, a + "\0", …))`, so other +/// attributes of the same entity — including prefix siblings such as `:ab` for +/// `:a` — and their version history are never resolved. +fn entity_attribute_facts( + d: &FactData, + entity_id: &EntityId, + attribute: &Attribute, +) -> Result> { + use crate::storage::index::EavtKey; + let matches = |f: &Fact| &f.entity == entity_id && &f.attribute == attribute; + + // Fallback: no indexes built yet + if d.pending_indexes.eavt.is_empty() && d.committed_index_reader.is_none() { + let mut result: Vec = d.facts.iter().filter(|f| matches(f)).cloned().collect(); + if let Some(loader) = &d.committed { + result.extend(loader.stream_all()?.into_iter().filter(|f| matches(f))); } + return Ok(result); + } + + let start = EavtKey { + entity: *entity_id, + attribute: attribute.clone(), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + // Exact successor of `attribute`: the smallest string that sorts after it is + // `attribute + "\0"`, so `[start, end)` holds exactly this (e, a) pair. Always + // valid UTF-8 — unlike incrementing the last byte, which fails for attributes + // ending in 0x7F or a character whose last UTF-8 byte is 0xBF (e.g. `:丿`). + let end = EavtKey { + entity: *entity_id, + attribute: format!("{attribute}\0"), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + + let mut facts = Vec::new(); + + // Pending: EAVT keys for the exact (e, a) are contiguous from `start`; + // the first key with a different entity or attribute ends the run. + for (key, &fr) in d.pending_indexes.eavt.range(start.clone()..) { + if key.entity != *entity_id || key.attribute != *attribute { + break; + } + facts.push(resolve_fact_ref(d, fr)?); + } - // Committed: on-disk B+tree range scan over exactly this attribute; the - // post-filter is a cheap guard, not something the range relies on. - if let Some(reader) = &d.committed_index_reader { - for fr in reader.range_scan_aevt(&start, Some(&end))? { - let fact = resolve_fact_ref(&d, fr)?; - if &fact.attribute == attribute { - facts.push(fact); - } + // Committed: on-disk B+tree range scan over exactly this (e, a) pair; the + // post-filter is a cheap guard, not something the range relies on. + if let Some(reader) = &d.committed_index_reader { + for fr in reader.range_scan_eavt(&start, Some(&end))? { + let fact = resolve_fact_ref(d, fr)?; + if matches(&fact) { + facts.push(fact); } } + } - Ok(facts) + Ok(facts) +} + +/// Get all facts for a specific attribute (index-driven). +fn attribute_facts(d: &FactData, attribute: &Attribute) -> Result> { + use crate::storage::index::AevtKey; + + // Fallback: no index + if d.pending_indexes.aevt.is_empty() && d.committed_index_reader.is_none() { + return Ok(all_facts(d)? + .into_iter() + .filter(|f| &f.attribute == attribute) + .collect()); + } + + let start = AevtKey { + attribute: attribute.clone(), + entity: uuid::Uuid::nil(), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + // Exact successor of `attribute` (see `get_facts_by_entity_attribute_indexed`): + // `[start, end)` holds exactly this attribute, never prefix siblings such as + // `:ab` for `:a`, and is valid UTF-8 for any attribute (#381). + let end = AevtKey { + attribute: format!("{attribute}\0"), + entity: uuid::Uuid::nil(), + valid_from: i64::MIN, + valid_to: i64::MIN, + tx_count: 0, + }; + + let mut facts = Vec::new(); + + // Pending: AEVT keys for the exact attribute are contiguous from `start`. + for (_, &fr) in d.pending_indexes.aevt.range(start.clone()..end.clone()) { + facts.push(resolve_fact_ref(d, fr)?); + } + + // Committed: on-disk B+tree range scan over exactly this attribute; the + // post-filter is a cheap guard, not something the range relies on. + if let Some(reader) = &d.committed_index_reader { + for fr in reader.range_scan_aevt(&start, Some(&end))? { + let fact = resolve_fact_ref(d, fr)?; + if &fact.attribute == attribute { + facts.push(fact); + } + } } + + Ok(facts) } /// Test-only helpers on FactStorage: for use in tests, not the production query path. @@ -1793,6 +1910,54 @@ mod tests { assert_eq!(storage.fact_count(), 2); } + #[test] + fn load_facts_appends_a_batch_and_skips_duplicates() { + let storage = FactStorage::new(); + let a = Fact::new( + uuid::Uuid::new_v4(), + ":t/a".to_string(), + Value::Integer(1), + 1, + ); + let b = Fact::new( + uuid::Uuid::new_v4(), + ":t/a".to_string(), + Value::Integer(2), + 1, + ); + assert_eq!( + storage + .load_facts(vec![a.clone(), b.clone(), a.clone()]) + .unwrap(), + 2 + ); + assert_eq!(storage.fact_count(), 2); + assert_eq!(storage.load_facts(vec![b]).unwrap(), 0); + assert_eq!(storage.fact_count(), 2); + } + + #[test] + fn facts_for_lookups_concatenates_each_lookup_in_order() { + let storage = FactStorage::new(); + let e = uuid::Uuid::new_v4(); + let (x, y) = (":t/x".to_string(), ":t/y".to_string()); + let fx = Fact::new(e, x.clone(), Value::Integer(1), 1); + let fy = Fact::new(e, y.clone(), Value::Integer(2), 1); + storage.load_facts(vec![fx, fy]).unwrap(); + let lookups = [ + Lookup::Attribute(&y), + Lookup::EntityAttribute(&e, &x), + Lookup::Entity(&e), + ]; + let attrs: Vec = storage + .get_facts_for_lookups(&lookups) + .unwrap() + .into_iter() + .map(|f| f.attribute) + .collect(); + assert_eq!(attrs, [":t/y", ":t/x", ":t/x", ":t/y"]); + } + #[test] fn test_load_fact_duplicate_detection_includes_asserted() { let storage = FactStorage::new(); diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 896629031..ead5d24c2 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -382,6 +382,7 @@ impl DatalogExecutor { /// Results are not deduplicated: the sole caller feeds them to `net_asserted_facts`, /// which is idempotent under duplicated records. fn selective_fact_fetch(&self, patterns: &[Pattern], threshold: usize) -> Option> { + use crate::graph::storage::Lookup; use std::collections::{BTreeMap, BTreeSet}; // Per bound entity: Some(attrs) = only these attributes are referenced; @@ -433,26 +434,19 @@ impl DatalogExecutor { return None; } - let mut all_facts: Vec = Vec::new(); - + // Every lookup reads one state of the store, so a commit cannot land + // between two patterns of this query (#555). + let mut lookups: Vec> = Vec::with_capacity(total); for (uid, attrs) in &entity_attrs { match attrs { Some(attrs) if narrow => { - for attr in attrs { - all_facts.extend( - self.storage - .get_facts_by_entity_attribute_indexed(uid, attr) - .ok()?, - ); - } + lookups.extend(attrs.iter().map(|a| Lookup::EntityAttribute(uid, a))); } - _ => all_facts.extend(self.storage.get_facts_by_entity(uid).ok()?), + _ => lookups.push(Lookup::Entity(uid)), } } - - for attr in &attributes { - all_facts.extend(self.storage.get_facts_by_attribute(attr).ok()?); - } + lookups.extend(attributes.iter().map(Lookup::Attribute)); + let all_facts = self.storage.get_facts_for_lookups(&lookups).ok()?; Some(all_facts) } diff --git a/src/storage/persistent_facts.rs b/src/storage/persistent_facts.rs index 47693af24..9eac695d7 100644 --- a/src/storage/persistent_facts.rs +++ b/src/storage/persistent_facts.rs @@ -870,8 +870,13 @@ impl PersistentFactStorage { return Ok(()); } + // Queries wait until the new state is published: the new fact pages + // overwrite the pages of the index they read (#558). Taken before the + // backend lock, the order queries use. + let mut store = self.storage.lock_for_checkpoint(); + // ── Step A: read current header + stream old B+tree entries BEFORE overwriting ── - let pending_facts = self.storage.get_pending_facts(); + let pending_facts = store.pending_facts(); let mut backend = self .backend .lock() @@ -1035,7 +1040,6 @@ impl PersistentFactStorage { committed_fact_pages: self.committed_fact_pages.clone(), first_fact_page: 1, }); - self.storage.set_committed_reader(loader); let index_reader: Arc = Arc::new(OnDiskIndexReader::new( @@ -1046,10 +1050,9 @@ impl PersistentFactStorage { avet_root, vaet_root, )); - self.storage.set_committed_index_reader(index_reader); - - // Clear pending — all data now on disk - self.storage.post_checkpoint_clear(); + // Readers in place and pending cleared — all data now on disk — in + // one step under the checkpoint lock. + store.publish(loader, index_reader); Ok(()) } diff --git a/tests/snapshot_atomicity_test.rs b/tests/snapshot_atomicity_test.rs new file mode 100644 index 000000000..3f992f624 --- /dev/null +++ b/tests/snapshot_atomicity_test.rs @@ -0,0 +1,210 @@ +//! A query must see each transaction whole (#554) and read one state of the +//! database for all its patterns (#555) while other threads commit. +//! +//! Each test runs a writer thread against a reader thread for a fixed time and +//! fails on the first answer that breaks the invariant. +#![cfg(not(target_arch = "wasm32"))] + +use minigraf::{Minigraf, OpenOptions, QueryResult}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::thread; +use std::time::{Duration, Instant}; + +const RUN_FOR: Duration = Duration::from_secs(3); + +fn rows(db: &Minigraf, query: &str) -> Vec> { + match db.execute(query).unwrap() { + QueryResult::QueryResults { results, .. } => results + .iter() + .map(|r| { + r.iter() + .map(|v| match v { + minigraf::Value::Integer(n) => n.to_string(), + minigraf::Value::Keyword(k) => k.clone(), + minigraf::Value::String(s) => s.clone(), + _ => "other".to_string(), + }) + .collect() + }) + .collect(), + _ => panic!("expected query results"), + } +} + +/// Runs `write` in a loop on one thread and `check` in a loop on this one +/// until `RUN_FOR` passes. `check` returns an error message on a violation. +fn race( + db: Arc, + mut write: impl FnMut(&Minigraf, u64) + Send + 'static, + mut check: impl FnMut(&Minigraf) -> Option, +) -> Option { + let stop = Arc::new(AtomicBool::new(false)); + let writer = { + let (db, stop) = (Arc::clone(&db), Arc::clone(&stop)); + thread::spawn(move || { + let mut i = 0; + while !stop.load(Ordering::Relaxed) { + write(&db, i); + i += 1; + } + }) + }; + let start = Instant::now(); + let mut found = None; + while start.elapsed() < RUN_FOR { + if let Some(msg) = check(&db) { + found = Some(msg); + break; + } + } + stop.store(true, Ordering::Relaxed); + writer.join().unwrap(); + found +} + +const N: u64 = 500; + +fn batch(i: u64) -> String { + let facts: String = (0..N) + .map(|j| format!("[:b{i}x{j} :batch/id {i}]")) + .collect(); + format!("(transact [{facts}])") +} + +/// Every write commits `N` facts in one transaction; a query must never see +/// a count that is not a multiple of `N` (#554). +fn commit_is_seen_whole(db: Minigraf, write: impl FnMut(&Minigraf, u64) + Send + 'static) { + let found = race(Arc::new(db), write, |db| { + let n = rows(db, "(query [:find ?e :where [?e :batch/id ?b]])").len() as u64; + (n % N != 0).then(|| format!("saw {n} facts, not a multiple of {N}")) + }); + if let Some(msg) = found { + panic!("partial transaction visible: {msg}"); + } +} + +fn write_transaction(db: &Minigraf, i: u64) { + let mut tx = db.begin_write().unwrap(); + tx.execute(&batch(i)).unwrap(); + tx.commit().unwrap(); +} + +fn execute_transact(db: &Minigraf, i: u64) { + db.execute(&batch(i)).unwrap(); +} + +fn file_db(dir: &tempfile::TempDir) -> Minigraf { + let opts = OpenOptions::default().wal_checkpoint_threshold(usize::MAX); + Minigraf::open_with_options(dir.path().join("w.graph"), opts).unwrap() +} + +#[test] +fn write_transaction_commit_is_seen_whole_in_memory() { + commit_is_seen_whole(Minigraf::in_memory().unwrap(), write_transaction); +} + +#[test] +fn write_transaction_commit_is_seen_whole_on_file() { + let dir = tempfile::tempdir().unwrap(); + commit_is_seen_whole(file_db(&dir), write_transaction); +} + +#[test] +fn execute_transact_is_seen_whole_in_memory() { + commit_is_seen_whole(Minigraf::in_memory().unwrap(), execute_transact); +} + +#[test] +fn execute_transact_is_seen_whole_on_file() { + let dir = tempfile::tempdir().unwrap(); + commit_is_seen_whole(file_db(&dir), execute_transact); +} + +/// A query fetched each attribute with its own `get_live_facts` call, each +/// under its own read lock, so a transaction that committed between the two +/// calls was seen by one pattern and not the other (#555). Every transaction +/// writes `:pair/x` and `:pair/y` for one entity together, atomically (#554). +/// Attributes are fetched in sorted order, so only the query whose positive +/// pattern is `:pair/y` can catch a commit between the fetches; both are run. +fn patterns_see_one_state(db: Minigraf) { + let found = race( + Arc::new(db), + |db, i| { + db.execute(&format!( + "(transact [[:p{i} :pair/x {i}] [:p{i} :pair/y {i}]])" + )) + .unwrap(); + }, + |db| { + for q in [ + "(query [:find ?e :where [?e :pair/x ?v] (not [?e :pair/y ?v])])", + "(query [:find ?e :where [?e :pair/y ?v] (not [?e :pair/x ?v])])", + ] { + let n = rows(db, q).len(); + if n != 0 { + return Some(format!("{n} entities with one attribute of the pair: {q}")); + } + } + None + }, + ); + if let Some(msg) = found { + panic!("one query saw two database states: {msg}"); + } +} + +#[test] +fn patterns_of_one_query_see_one_state_in_memory() { + patterns_see_one_state(Minigraf::in_memory().unwrap()); +} + +#[test] +fn patterns_of_one_query_see_one_state_on_file() { + let dir = tempfile::tempdir().unwrap(); + // No checkpoints: `checkpoint_never_hides_committed_facts` covers those. + patterns_see_one_state(file_db(&dir)); +} + +/// A checkpoint wrote its new fact pages over the index pages queries were +/// reading, and published its new readers in three locked steps, so a query +/// during a checkpoint could miss committed facts or see half a transaction +/// (#558). Every write here is followed by a checkpoint. +#[test] +fn checkpoint_never_hides_committed_facts() { + let dir = tempfile::tempdir().unwrap(); + let committed = Arc::new(AtomicU64::new(0)); + let found = { + let written = Arc::clone(&committed); + race( + Arc::new(file_db(&dir)), + move |db, i| { + db.execute(&format!( + "(transact [[:c{i} :pair/x {i}] [:c{i} :pair/y {i}]])" + )) + .unwrap(); + written.store(i + 1, Ordering::SeqCst); + db.checkpoint().unwrap(); + }, + |db| { + let before = committed.load(Ordering::SeqCst) as usize; + let half = rows( + db, + "(query [:find ?e :where [?e :pair/y ?v] (not [?e :pair/x ?v])])", + ) + .len(); + let seen = rows(db, "(query [:find ?e :where [?e :pair/x ?v]])").len(); + if half != 0 { + Some(format!("{half} entities with y and no x")) + } else if seen < before { + Some(format!("{seen} entities seen, {before} committed before")) + } else { + None + } + }, + ) + }; + if let Some(msg) = found { + panic!("a query during a checkpoint saw the wrong state: {msg}"); + } +}