fix: gate read-only Graph Drop and flush against on-disk writes

db4-graph/src/lib.rs

@@ -157,7 +157,13 @@ where
 
     fn load_inner(path: impl AsRef<Path>, ext: EXT, read_only: bool) -> Result<Self, StorageError> {
         let path = path.as_ref();
-        let storage = Layer::load(path, ext)?;
+        // Load the storage layer as read-only when requested so its Drop
+        // and flush paths do not write to the shared graph directory.
+        let storage = if read_only {
+            Layer::load_read_only(path, ext)?
+        } else {
+            Layer::load(path, ext)?
+        };
         let id_type = storage.nodes().id_type();
 
         let gid_resolver_dir = path.join("gid_resolver");

db4-storage/src/lib.rs

@@ -127,6 +127,9 @@ pub mod error {
 
         #[error("Graph requires WAL recovery; load with WAL enabled")]
         RecoveryRequired,
+
+        #[error("Graph is read-only")]
+        ReadOnly,
     }
 
     impl StorageError {

db4-storage/src/pages/mod.rs

@@ -57,6 +57,7 @@ pub struct GraphStore<
     graph_dir: Option<PathBuf>,
     event_id: AtomicUsize,
     ext: EXT,
+    read_only: bool,
 }
 
 impl<
@@ -67,6 +68,10 @@ impl<
 > GraphStore<NS, ES, GS, EXT>
 {
     pub fn flush(&self) -> Result<(), StorageError> {
+        if self.read_only {
+            return Err(StorageError::ReadOnly);
+        }
+
         let node_types = self.nodes.prop_meta().get_all_node_types();
         let config = self.ext.config().with_node_types(node_types);
 
@@ -153,6 +158,7 @@ impl<
             event_id: AtomicUsize::new(0),
             graph_dir: graph_dir.map(|p| p.to_path_buf()),
             ext,
+            read_only: false,
         }
     }
 
@@ -191,9 +197,20 @@ impl<
             event_id: AtomicUsize::new(t_len),
             graph_dir: Some(graph_dir.as_ref().to_path_buf()),
             ext,
+            read_only: false,
         })
     }
 
+    /// Load a graph directory as a read-only snapshot.  The returned store
+    /// is byte-identical to `load` at the storage level, but its Drop and
+    /// `flush` paths are guarded to never write to the graph directory —
+    /// safe to attach to a directory a live writer is holding open.
+    pub fn load_read_only(graph_dir: impl AsRef<Path>, ext: EXT) -> Result<Self, StorageError> {
+        let mut store = Self::load(graph_dir, ext)?;
+        store.read_only = true;
+        Ok(store)
+    }
+
     pub fn read_locked(self: &Arc<Self>) -> ReadLockedGraphStore<NS, ES, GS, EXT> {
         let nodes = self.nodes.locked().into();
         let edges = self.edges.locked().into();
@@ -360,6 +377,14 @@ impl<
 > Drop for GraphStore<NS, ES, GS, EXT>
 {
     fn drop(&mut self) {
+        // A read-only handle must never write to the graph directory —
+        // running the clean-shutdown protocol would append a shutdown
+        // checkpoint to the shared WAL and rewrite the control file,
+        // corrupting a concurrent writer's crash recovery.
+        if self.read_only {
+            return;
+        }
+
         let wal = self.ext.wal();
         let control_file = self.ext.control_file();
 

python/tests/test_base_install/test_graphql/test_read_only_durability.py

@@ -0,0 +1,221 @@
+"""
+Durability regression tests for concurrent read-only opens.
+
+There are three distinct bugs in this area, targeted by three separate
+tests below:
+
+  Bug #1  Reader Drop appends a shutdown checkpoint to the writer's WAL,
+          overwriting live records at LSN 256 of ``log.0``.  On next
+          reopen after a writer crash, recovery fails with
+          "Expected checkpoint at given LSN".
+  Bug #2  Reader Drop rewrites ``.meta`` via a fixed-name ``.tmp`` file.
+          Racing renames intermittently ENOENT at ``meta_file.rs:80``.
+  Bug #3  Reader ``Graph.load`` races the writer's segment-file
+          creation and can list a segment directory before its layer
+          stats file has been written, causing ENOENT at
+          ``disk_layer_stats/mod.rs:301``.
+
+Bugs #1 and #2 are Drop side-effects on shared state.  Bug #3 is a
+reader-open race, independent of Drop.
+
+See ``docs/db-v4-wal-explainer.md`` for the underlying mechanism.
+"""
+
+import hashlib
+import os
+import signal
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+
+pytestmark = pytest.mark.skipif(
+    "DISK_TEST_MARK" not in os.environ,
+    reason="disk-backed graph tests require the storage feature",
+)
+
+
+# Subprocess writer: opens the graph at argv[1] and appends nodes + flushes
+# in a tight loop.  Prints ``UP`` to stdout so the parent knows the writer
+# has taken the writer lock and started producing data.
+_WRITER_SCRIPT = """
+import sys, time
+from raphtory import Graph
+g = Graph(sys.argv[1])
+print("UP", flush=True)
+i = 0
+while True:
+    for _ in range(20):
+        g.add_node(1 + i, f"n{i}", properties={"v": i})
+        i += 1
+    g.flush()
+    time.sleep(0.01)
+"""
+
+
+def _spawn_writer(path):
+    p = subprocess.Popen(
+        [sys.executable, "-c", _WRITER_SCRIPT, path],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    up = p.stdout.readline()
+    assert up.strip() == "UP", f"writer failed to start: {up!r}"
+    # Give the writer a moment to accumulate real data before observing it —
+    # otherwise the reader races the very first flush.
+    time.sleep(0.2)
+    return p
+
+
+def _kill(p):
+    p.send_signal(signal.SIGKILL)
+    p.wait(timeout=10)
+
+
+def _hash_all_files(root):
+    """Return {relative_path: sha256_hex} for every file under `root`.
+    Any change to any file shows up as a differing dict value."""
+    root = Path(root)
+    return {
+        str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest()
+        for p in sorted(root.rglob("*"))
+        if p.is_file()
+    }
+
+
+# --- Test 1: CONTROL --------------------------------------------------------
+
+
+def test_writer_crash_recovers_cleanly_without_readers(tmp_path):
+    """CONTROL: a writer that is SIGKILLed with no concurrent readers must
+    still be recoverable on reopen.  Ordinary crash recovery must work
+    before we can meaningfully assert anything about the concurrent-reader
+    cases below."""
+    from raphtory import Graph
+
+    path = str(tmp_path / "g")
+    w = _spawn_writer(path)
+    try:
+        time.sleep(1.5)
+    finally:
+        _kill(w)
+
+    g = Graph.load(path)
+    assert g.count_nodes() > 0
+
+
+# --- Test 2: Bugs #1 and #2 (deterministic Drop side-effect test) ----------
+
+
+def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path):
+    """Bugs #1 and #2: a read-only ``Graph.load(...)`` followed by drop
+    must not modify any file in the graph directory.
+
+    We freeze the writer subprocess with ``SIGSTOP`` so it cannot be the
+    source of any file change during our observation window.  Any diff
+    between the before/after directory hashes is therefore caused by the
+    read-only handle itself.
+
+    Before the fix this catches:
+
+    * Bug #1 - Drop for GraphStore appends a shutdown checkpoint to the
+      WAL, changing ``wal/logs/log.0`` bytes.
+    * Bug #2 - Drop for Storage rewrites ``.meta``, changing its bytes
+      (and racing on ``.tmp`` under concurrency).
+    """
+    from raphtory import Graph
+
+    path = Path(tmp_path) / "g"
+    w = _spawn_writer(str(path))
+    # Give the writer time to produce enough data that log.0 has real
+    # records past the header (which is what Bug #1 clobbers).
+    time.sleep(0.5)
+
+    os.kill(w.pid, signal.SIGSTOP)
+    try:
+        before = _hash_all_files(path)
+
+        rg = Graph.load(str(path), read_only=True)
+        rg.count_nodes()
+        del rg
+
+        after = _hash_all_files(path)
+    finally:
+        # Resume the writer so it can be cleanly SIGKILLed.
+        os.kill(w.pid, signal.SIGCONT)
+        _kill(w)
+
+    diffs = {
+        k: (before.get(k), after.get(k))
+        for k in before.keys() | after.keys()
+        if before.get(k) != after.get(k)
+    }
+    assert not diffs, (
+        f"read-only open+drop modified {len(diffs)} file(s) in the graph "
+        f"directory: {sorted(diffs.keys())}"
+    )
+
+
+# --- Test 3: Bug #3 (reader-open race, xfail placeholder) ------------------
+
+
+@pytest.mark.xfail(
+    reason=(
+        "Bug #3: reader Graph.load races the writer's segment-file creation; "
+        "reader can list a segment directory before its layer stats file has "
+        "been written, causing ENOENT at disk_layer_stats/mod.rs:301. "
+        "Separate follow-up from the Drop-side-effect fix (Bugs #1 and #2)."
+    ),
+    strict=False,
+)
+def test_reader_open_does_not_race_writer_segment_creation(tmp_path):
+    """Bug #3: while a writer is writing + flushing continuously, reader
+    threads that repeatedly open the graph read-only should not encounter
+    ENOENT on layer stats files that the writer is mid-creating.
+
+    Marked ``xfail`` because the fix is a separate change — the writer's
+    segment creation needs to become atomic before this can pass.
+    """
+    from raphtory import Graph
+
+    path = str(tmp_path / "g")
+    w = _spawn_writer(path)
+
+    stop = threading.Event()
+    enoent_errors = []
+    lock = threading.Lock()
+
+    def reader():
+        while not stop.is_set():
+            try:
+                rg = Graph.load(path, read_only=True)
+                rg.count_nodes()
+                del rg
+            except Exception as exc:  # noqa: BLE001 - collect any error
+                msg = str(exc)
+                # Focus on the specific race we're tracking here — ignore
+                # any other transient errors that a general stress test
+                # might surface.
+                if "disk_layer_stats" in msg and "No such file" in msg:
+                    with lock:
+                        enoent_errors.append(repr(exc))
+
+    threads = [threading.Thread(target=reader, daemon=True) for _ in range(4)]
+    try:
+        for t in threads:
+            t.start()
+        time.sleep(1.5)
+        stop.set()
+        for t in threads:
+            t.join(timeout=5)
+    finally:
+        _kill(w)
+
+    assert not enoent_errors, (
+        f"reader-open raced writer segment creation: {len(enoent_errors)} "
+        f"ENOENT errors, first: {enoent_errors[0]}"
+    )

raphtory/src/db/api/storage/storage.rs

@@ -62,13 +62,22 @@ pub use storage::{
 #[derive(Debug, Default)]
 pub struct Storage {
     graph: GraphStorage,
+    // Set by `load_read_only_*`.  Guards Drop so a read-only handle does
+    // not rewrite the on-disk `.meta` file.
+    read_only: bool,
     #[cfg(feature = "search")]
     pub(crate) index: RwLock<GraphIndex>,
 }
 
 #[cfg(feature = "io")]
 impl Drop for Storage {
     fn drop(&mut self) {
+        // A read-only handle must never write to the graph directory,
+        // including the `.meta` counts file (a fixed-name `.tmp` collides
+        // with the writer's own `.meta` refresh under concurrency).
+        if self.read_only {
+            return;
+        }
         if let Some(disk_path) = self.graph.disk_storage_path() {
             let disk_path = disk_path.to_path_buf();
             let node_count = self.graph.unfiltered_num_nodes(&LayerIds::All);
@@ -120,6 +129,7 @@ impl Storage {
 
         Ok(Self {
             graph: GraphStorage::Unlocked(Arc::new(temporal_graph)),
+            read_only: false,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         })
@@ -130,6 +140,7 @@ impl Storage {
         let temporal_graph = TemporalGraph::new(ext)?;
         Ok(Self {
             graph: GraphStorage::Unlocked(Arc::new(temporal_graph)),
+            read_only: false,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         })
@@ -144,6 +155,7 @@ impl Storage {
 
         Ok(Self {
             graph: GraphStorage::Unlocked(Arc::new(temporal_graph)),
+            read_only: false,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         })
@@ -157,6 +169,7 @@ impl Storage {
 
         Ok(Self {
             graph: GraphStorage::Unlocked(Arc::new(temporal_graph)),
+            read_only: false,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         })
@@ -174,6 +187,7 @@ impl Storage {
 
         Ok(Self {
             graph: GraphStorage::Mem(locked),
+            read_only: true,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         })
@@ -216,6 +230,7 @@ impl Storage {
     pub(crate) fn from_inner(graph: GraphStorage) -> Self {
         Self {
             graph,
+            read_only: false,
             #[cfg(feature = "search")]
             index: RwLock::new(GraphIndex::Empty),
         }
@@ -229,6 +244,10 @@ impl Storage {
     pub fn read_only(&self) -> Self {
         Self {
             graph: self.graph.lock(),
+            // The original Storage still owns the writer's Drop protocol
+            // for the underlying graph; this in-process view must not
+            // duplicate the on-disk `.meta` refresh on its own drop.
+            read_only: true,
             #[cfg(feature = "search")]
             index: RwLock::new(self.index.read().clone()),
         }