adityamukho · Oct 11, 2026 · Oct 11, 2026 · Oct 11, 2026 · Oct 11, 2026 · Oct 11, 2026
@@ -49,6 +49,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Reopening next to a WAL whose entries were all already checkpointed no longer rewinds the transaction counter, so later transactions no longer reuse `tx_count` values and `:as-of` history stays correct (#447; also fixed in v2.0.3).
 - `checkpoint()` removes a WAL whose entries are all already in the file (#457). When it had no facts to flush, it returned early and left the WAL until a later write was checkpointed. That WAL is left by a crash between a checkpoint's commit and its WAL delete, or by a failed delete (`WAL-006`). Its contents were always skipped on replay, so no data was affected; the sidecar file just stayed. Also fixed on the v2.x line (`main`, #540), not yet released.
 - Closing a handle opened with `wal_checkpoint_threshold(usize::MAX)` no longer writes its pending facts to the `.graph` file (#476). The sentinel was documented to suppress the close-time checkpoint, and `Minigraf`'s close did skip it, but the storage layer still saved on drop, leaving a WAL whose entries were all checkpointed. The file now stays as the last explicit `checkpoint()` left it, and the WAL keeps every pending fact for the next open to replay, so bulk loaders and benchmarks see the pending WAL they ask for. No data is lost either way. Also fixed on the v2.x line (`main`, #540), not yet released.
+- **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.
+- **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 (an entity with one attribute but not another that was always written with it). The lookups now run under one lock. Stored data was never affected. Affected every release since v1.1.0.
 
 ### Internal
 

@@ -201,7 +201,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are
 
 ## Test Coverage
 
-**1432 tests** (1422 passing, 10 ignored; unit + integration + doc).
+**1441 tests** (1430 passing, 11 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.

@@ -131,7 +131,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit
 
 ```bash
 cargo run          # interactive Datalog REPL
-cargo test         # run 1432 tests
+cargo test         # run 1441 tests
 cargo run < demos/demo_recursive.txt   # recursive rules demo
 ```
 

@@ -343,12 +343,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)
+                .map_err(|e| JsValue::from_str(&e.to_string()))?;
 
             inner.pfs.mark_dirty();
             inner

@@ -630,9 +630,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;
@@ -756,10 +754,9 @@ 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)?;
 
             // Trigger auto-checkpoint AFTER facts are in FactStorage so the
             // checkpoint captures the newly written facts. An open fact log
@@ -1591,10 +1588,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. An open fact log

@@ -309,6 +309,29 @@ 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<Item = Fact>) -> Result<usize> {
+        let mut d = self
+            .data
+            .write()
+            .map_err(|_| err_coded!(ErrorCode::Int050, "data"))?;
+        let mut loaded = 0usize;
+        for fact in facts {
+            if !d.pending_keys.insert(pending_key(&fact)) {
+                continue;
+            }
+            let pos = d.facts.len();
+            d.pending_indexes.insert(&fact, pos);
+            d.facts.push(fact);
+            loaded = loaded.saturating_add(1);
+        }
+        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
@@ -666,63 +689,90 @@ impl FactStorage {
         scan: crate::storage::Scan
10000
<'_>,
         as_of: Option<u64>,
     ) -> Result<Vec<Fact>> {
-        use crate::storage::Scan;
-        use crate::storage::index::{AevtKey, EavtKey};
+        self.get_live_facts_of_scans(&[scan], as_of)
+    }
+
+    /// [`Self::get_live_facts`] for several scans, all read from one state of
+    /// the store: no commit lands between two of them (#555). The results are
+    /// concatenated in `scans` order and not deduplicated.
+    pub(crate) fn get_live_facts_of_scans(
+        &self,
+        scans: &[crate::storage::Scan<'_>],
+        as_of: Option<u64>,
+    ) -> Result<Vec<Fact>> {
         let d = self.data.read().unwrap_or_else(|e| e.into_inner());
         let mut facts = Vec::new();
-        let mut push = |pos: usize| -> Result<()> {
-            let f = pending_fact(&d, pos)?;
-            if as_of.is_none_or(|n| f.tx_count <= n) {
-                facts.push(f);
-            }
-            Ok(())
-        };
-        match scan {
-            Scan::All => {
-                for pos in 0..d.facts.len() {
-                    push(pos)?;
-                }
+        for &scan in scans {
+            live_facts_of_scan(&d, scan, as_of, &mut facts)?;
+        }
+        Ok(facts)
+    }
+}
+
+/// The pending facts of `scan` up to `as_of`, then its committed live facts,
+/// appended to `facts`. The caller holds `d`'s lock for the whole read.
+fn live_facts_of_scan(
+    d: &FactData,
+    scan: crate::storage::Scan<'_>,
+    as_of: Option<u64>,
+    facts: &mut Vec<Fact>,
+) -> Result<()> {
+    use crate::storage::Scan;
+    use crate::storage::index::{AevtKey, EavtKey};
+    let mut push = |pos: usize| -> Result<()> {
+        let f = pending_fact(d, pos)?;
+        if as_of.is_none_or(|n| f.tx_count <= n) {
+            facts.push(f);
+        }
+        Ok(())
+    };
+    match scan {
+        Scan::All => {
+            for pos in 0..d.facts.len() {
+                push(pos)?;
             }
-            Scan::Entity(entity) => {
-                for (key, &pos) in d
-                    .pending_indexes
-                    .eavt
-                    .range(EavtKey::entity_start(*entity)..)
-                {
-                    if key.entity != *entity {
-                        break;
-                    }
-                    push(pos)?;
+        }
+        Scan::Entity(entity) => {
+            for (key, &pos) in d
+                .pending_indexes
+                .eavt
+                .range(EavtKey::entity_start(*entity)..)
+            {
+                if key.entity != *entity {
+                    break;
                 }
+                push(pos)?;
             }
-            Scan::EntityAttribute(entity, attribute) => {
-                let start = EavtKey::entity_attribute_start(*entity, attribute);
-                for (key, &pos) in d.pending_indexes.eavt.range(start..) {
-                    if key.entity != *entity || key.attribute != attribute {
-                        break;
-                    }
-                    push(pos)?;
+        }
+        Scan::EntityAttribute(entity, attribute) => {
+            let start = EavtKey::entity_attribute_start(*entity, attribute);
+            for (key, &pos) in d.pending_indexes.eavt.range(start..) {
+                if key.entity != *entity || key.attribute != attribute {
+                    break;
                 }
+                push(pos)?;
             }
-            Scan::Attribute(attribute) => {
-                for (key, &pos) in d
-                    .pending_indexes
-                    .aevt
-                    .range(AevtKey::attribute_start(attribute)..)
-                {
-                    if key.attribute != attribute {
-                        break;
-                    }
-                    push(pos)?;
+        }
+        Scan::Attribute(attribute) => {
+            for (key, &pos) in d
+                .pending_indexes
+                .aevt
+                .range(AevtKey::attribute_start(attribute)..)
+            {
+                if key.attribute != attribute {
+                    break;
                 }
+                push(pos)?;
             }
         }
-        if let Some(reader) = &d.committed {
-            facts.extend(reader.live_facts(scan, as_of)?);
-        }
-        Ok(facts)
     }
+    if let Some(reader) = &d.committed {
+        facts.extend(reader.live_facts(scan, as_of)?);
+    }
+    Ok(())
+}
 
+impl FactStorage {
     /// History read (every record); tests compare it with live reads and models.
     #[cfg(test)]
     /// Get all facts for a specific entity (index-driven).
@@ -1641,6 +1691,54 @@ mod tests {
         assert_eq!(storage.fact_count(), 2);
     }
 
+    #[test]
+    fn live_facts_of_scans_concatenates_each_scan_in_order() {
+        use crate::storage::Scan;
+        let storage = FactStorage::new();
+        let e = uuid::Uuid::new_v4();
+        let x = Fact::new(e, ":t/x".to_string(), Value::Integer(1), 1);
+        let y = Fact::new(e, ":t/y".to_string(), Value::Integer(2), 1);
+        storage.load_facts(vec![x, y]).unwrap();
+        let scans = [
+            Scan::Attribute(":t/y"),
+            Scan::Attribute(":t/x"),
+            Scan::Entity(&e),
+        ];
+        let attrs: Vec<String> = storage
+            .get_live_facts_of_scans(&scans, None)
+            .unwrap()
+            .into_iter()
+            .map(|f| f.attribute)
+            .collect();
+        assert_eq!(attrs, [":t/y", ":t/x", ":t/x", ":t/y"]);
+    }
+
+    #[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 test_load_fact_duplicate_detection_includes_asserted() {
         let storage = FactStorage::new();

@@ -383,7 +383,8 @@ impl DatalogExecutor {
     ///
     /// Results are not deduplicated: the sole caller feeds them to `net_asserted_facts`,
     /// which is idempotent under duplicated records. Each lookup returns the pending facts
-    /// up to `as_of` and the committed facts live at `as_of` (`FactStorage::get_live_facts`).
+    /// up to `as_of` and the committed facts live at `as_of`, and all lookups read one
+    /// state of the store (`FactStorage::get_live_facts_of_scans`, #555).
     fn selective_fact_fetch(
         &self,
         patterns: &[Pattern],
@@ -442,25 +443,19 @@ impl DatalogExecutor {
             return Ok(None);
         }
 
-        let mut all_facts: Vec<Fact> = Vec::new();
-
+        // Every scan reads one state of the store, so a commit cannot land
+        // between two patterns of this query (#555).
+        let mut scans: Vec<Scan<'_>> = 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_live_facts(Scan::EntityAttribute(uid, attr), as_of)?,
-                        );
-                    }
+                    scans.extend(attrs.iter().map(|a| Scan::EntityAttribute(uid, a)));
                 }
-                _ => all_facts.extend(self.storage.get_live_facts(Scan::Entity(uid), as_of)?),
+                _ => scans.push(Scan::Entity(uid)),
             }
         }
-
-        for attr in &attributes {
-            all_facts.extend(self.storage.get_live_facts(Scan::Attribute(attr), as_of)?);
-        }
+        scans.extend(attributes.iter().map(|a| Scan::Attribute(a)));
+        let all_facts = self.storage.get_live_facts_of_scans(&scans, as_of)?;
 
         Ok(Some(all_facts))
     }

@@ -163,9 +163,7 @@ impl<B: StorageBackend + 'static> PersistentFactStorage<B> {
         let max_tx = max_fact_tx.max(header.last_checkpointed_tx_count);
         // Pending facts are kept in tx order (the fact log relies on it).
         facts.sort_by_key(|f| f.tx_count);
-        for fact in facts {
-            let _ = self.storage.load_fact(fact)?;
-        }
+        let _ = self.storage.load_facts(facts)?;
         Ok(MetaPage {
             last_checkpointed_tx_count: max_tx,
             ..MetaPage::empty(1)