From 6d474cc745d46d67c84db71b0994ddb67ba8633a Mon Sep 17 00:00:00 2001 From: shivamka1 <4599890+shivamka1@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:50:33 +0100 Subject: [PATCH 1/6] fix: gate read-only Graph Drop and flush against on-disk writes --- db4-graph/src/lib.rs | 8 +- db4-storage/src/lib.rs | 3 + db4-storage/src/pages/mod.rs | 25 ++ .../test_graphql/test_read_only_durability.py | 221 ++++++++++++++++++ raphtory/src/db/api/storage/storage.rs | 19 ++ 5 files changed, 275 insertions(+), 1 deletion(-) create mode 100644 python/tests/test_base_install/test_graphql/test_read_only_durability.py diff --git a/db4-graph/src/lib.rs b/db4-graph/src/lib.rs index a5c665eb82..9b03072e80 100644 --- a/db4-graph/src/lib.rs +++ b/db4-graph/src/lib.rs @@ -157,7 +157,13 @@ where fn load_inner(path: impl AsRef, ext: EXT, read_only: bool) -> Result { 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"); diff --git a/db4-storage/src/lib.rs b/db4-storage/src/lib.rs index d5bc9618ba..97b567fdea 100644 --- a/db4-storage/src/lib.rs +++ b/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 { diff --git a/db4-storage/src/pages/mod.rs b/db4-storage/src/pages/mod.rs index 980b24ec1d..0bb8354817 100644 --- a/db4-storage/src/pages/mod.rs +++ b/db4-storage/src/pages/mod.rs @@ -57,6 +57,7 @@ pub struct GraphStore< graph_dir: Option, event_id: AtomicUsize, ext: EXT, + read_only: bool, } impl< @@ -67,6 +68,10 @@ impl< > GraphStore { 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, ext: EXT) -> Result { + let mut store = Self::load(graph_dir, ext)?; + store.read_only = true; + Ok(store) + } + pub fn read_locked(self: &Arc) -> ReadLockedGraphStore { let nodes = self.nodes.locked().into(); let edges = self.edges.locked().into(); @@ -360,6 +377,14 @@ impl< > Drop for GraphStore { 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(); diff --git a/python/tests/test_base_install/test_graphql/test_read_only_durability.py b/python/tests/test_base_install/test_graphql/test_read_only_durability.py new file mode 100644 index 0000000000..24f67e8dd5 --- /dev/null +++ b/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]}" + ) diff --git a/raphtory/src/db/api/storage/storage.rs b/raphtory/src/db/api/storage/storage.rs index bc5e8b7862..bc2e854b3e 100644 --- a/raphtory/src/db/api/storage/storage.rs +++ b/raphtory/src/db/api/storage/storage.rs @@ -62,6 +62,9 @@ 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, } @@ -69,6 +72,12 @@ pub struct Storage { #[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()), } From 13d2ae96df2c43fa56578421c0978b439e07b768 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 1 Jul 2026 17:34:27 +0000 Subject: [PATCH 2/6] chore: apply tidy-public auto-fixes --- .../test_base_install/test_graphql/test_read_only_durability.py | 1 - 1 file changed, 1 deletion(-) diff --git a/python/tests/test_base_install/test_graphql/test_read_only_durability.py b/python/tests/test_base_install/test_graphql/test_read_only_durability.py index 24f67e8dd5..e603d5860c 100644 --- a/python/tests/test_base_install/test_graphql/test_read_only_durability.py +++ b/python/tests/test_base_install/test_graphql/test_read_only_durability.py @@ -32,7 +32,6 @@ import pytest - pytestmark = pytest.mark.skipif( "DISK_TEST_MARK" not in os.environ, reason="disk-backed graph tests require the storage feature", From d3b25b1f1b47a6a1e780eb63275fbbf472ad9fd6 Mon Sep 17 00:00:00 2001 From: shivamka1 <4599890+shivamka1@users.noreply.github.com> Date: Thu, 2 Jul 2026 13:30:18 +0100 Subject: [PATCH 3/6] refactor shutdown guard --- db4-storage/src/pages/mod.rs | 247 ++++++++++++++++--------- raphtory/src/db/api/storage/storage.rs | 88 ++++++--- 2 files changed, 213 insertions(+), 122 deletions(-) diff --git a/db4-storage/src/pages/mod.rs b/db4-storage/src/pages/mod.rs index 0bb8354817..329150f0cb 100644 --- a/db4-storage/src/pages/mod.rs +++ b/db4-storage/src/pages/mod.rs @@ -57,7 +57,32 @@ pub struct GraphStore< graph_dir: Option, event_id: AtomicUsize, ext: EXT, - read_only: bool, + // `Some` on writable handles, `None` on read-only handles. On drop + // the guard runs the clean-shutdown protocol (flush + WAL shutdown + // checkpoint + control-file update); a `None` here means the drop + // does nothing, so a read-only handle can never write to the graph + // directory. Also gates `flush()`. + shutdown_guard: Option>, +} + +/// Owns the writer's clean-shutdown protocol. Constructed alongside a +/// writable `GraphStore`; dropped when that store is dropped, at which +/// point (if the DB is still `Running`) it flushes the segments and +/// records a shutdown checkpoint in the WAL. Read-only handles never +/// hold one, so they are structurally incapable of running this +/// protocol. +#[derive(Debug)] +pub struct WriterShutdownGuard< + NS: NodeSegmentOps, + ES: EdgeSegmentOps, + GS: GraphPropSegmentOps, + EXT: PersistenceStrategy, +> { + nodes: Arc>, + edges: Arc>, + graph_props: Arc>, + graph_dir: Option, + ext: EXT, } impl< @@ -65,13 +90,9 @@ impl< ES: EdgeSegmentOps, GS: GraphPropSegmentOps, EXT: PersistenceStrategy, -> GraphStore +> WriterShutdownGuard { - pub fn flush(&self) -> Result<(), StorageError> { - if self.read_only { - return Err(StorageError::ReadOnly); - } - + fn flush(&self) -> Result<(), StorageError> { let node_types = self.nodes.prop_meta().get_all_node_types(); let config = self.ext.config().with_node_types(node_types); @@ -86,6 +107,85 @@ impl< } } +impl< + NS: NodeSegmentOps, + ES: EdgeSegmentOps, + GS: GraphPropSegmentOps, + EXT: PersistenceStrategy, +> Drop for WriterShutdownGuard +{ + fn drop(&mut self) { + let wal = self.ext.wal(); + let control_file = self.ext.control_file(); + + // Skip running a clean flush if the DB is in shutdown or crash recovery. + // Note that the state can be Shutdown after load is called and before recovery is complete. + // A clean flush is performed even when state is WalDisabled or NotSupported. + if matches!( + control_file.db_state(), + DBState::Shutdown | DBState::CrashRecovery + ) { + return; + } + + match self.flush() { + Ok(_) => { + // Log a checkpoint record in the WAL, indicating that the DB was shutdown + // with all the segments flushed to disk. + // On startup, recovery is skipped since there are no pending writes to replay. + let checkpoint_lsn = match wal.log_shutdown_checkpoint() { + Ok(lsn) => lsn, + Err(err) => { + drop_error!("Failed to log shutdown checkpoint in drop: {err}"); + // this is unreachable with panic-on-drop + #[allow(unreachable_code)] + return; + } + }; + + // Flush up to the end of the WAL stream. + let flush_lsn = wal.position(); + + if let Err(err) = wal.flush(flush_lsn) { + drop_error!("Failed to flush checkpoint record in drop: {err}"); + // this is unreachable with panic-on-drop + #[allow(unreachable_code)] + return; + } + + // Record the checkpoint and shutdown state and write control file to disk. + control_file.set_checkpoint(checkpoint_lsn); + control_file.set_db_state(DBState::Shutdown); + + if let Err(err) = control_file.save() { + drop_error!("Failed to save control file in drop: {err}"); + // this is unreachable with panic-on-drop + #[allow(unreachable_code)] + return; + } + } + Err(err) => { + drop_error!("Failed to flush storage in drop: {err}"); + } + } + } +} + +impl< + NS: NodeSegmentOps, + ES: EdgeSegmentOps, + GS: GraphPropSegmentOps, + EXT: PersistenceStrategy, +> GraphStore +{ + pub fn flush(&self) -> Result<(), StorageError> { + match &self.shutdown_guard { + None => Err(StorageError::ReadOnly), + Some(guard) => guard.flush(), + } + } +} + #[derive(Debug)] pub struct ReadLockedGraphStore< NS: NodeSegmentOps, @@ -151,18 +251,43 @@ impl< .expect("Failed to write config to disk"); } + let graph_dir = graph_dir.map(|p| p.to_path_buf()); + let shutdown_guard = Some(WriterShutdownGuard { + nodes: node_storage.clone(), + edges: edge_storage.clone(), + graph_props: graph_prop_storage.clone(), + graph_dir: graph_dir.clone(), + ext: ext.clone(), + }); + Self { nodes: node_storage, edges: edge_storage, graph_props: graph_prop_storage, event_id: AtomicUsize::new(0), - graph_dir: graph_dir.map(|p| p.to_path_buf()), + graph_dir, ext, - read_only: false, + shutdown_guard, } } pub fn load(graph_dir: impl AsRef, ext: EXT) -> Result { + Self::load_impl(graph_dir, ext, /* read_only */ false) + } + + /// Load a graph directory as a read-only snapshot. The returned store + /// omits the writer's shutdown guard, so its Drop and `flush` paths + /// are structurally incapable of writing 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, ext: EXT) -> Result { + Self::load_impl(graph_dir, ext, /* read_only */ true) + } + + fn load_impl( + graph_dir: impl AsRef, + ext: EXT, + read_only: bool, + ) -> Result { let nodes_path = graph_dir.as_ref().join("nodes"); let edges_path = graph_dir.as_ref().join("edges"); let graph_props_path = graph_dir.as_ref().join("graph_props"); @@ -190,27 +315,33 @@ impl< + node_storage.t_len() + graph_prop_storage.segment().num_updates(); + let graph_dir = Some(graph_dir.as_ref().to_path_buf()); + // The guard is only constructed for writable handles. Constructing + // it and then setting the field to `None` would run its Drop and + // execute the very shutdown protocol we're trying to avoid. + let shutdown_guard = if read_only { + None + } else { + Some(WriterShutdownGuard { + nodes: node_storage.clone(), + edges: edge_storage.clone(), + graph_props: graph_prop_storage.clone(), + graph_dir: graph_dir.clone(), + ext: ext.clone(), + }) + }; + Ok(Self { nodes: node_storage, edges: edge_storage, graph_props: graph_prop_storage, event_id: AtomicUsize::new(t_len), - graph_dir: Some(graph_dir.as_ref().to_path_buf()), + graph_dir, ext, - read_only: false, + shutdown_guard, }) } - /// 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, ext: EXT) -> Result { - let mut store = Self::load(graph_dir, ext)?; - store.read_only = true; - Ok(store) - } - pub fn read_locked(self: &Arc) -> ReadLockedGraphStore { let nodes = self.nodes.locked().into(); let edges = self.edges.locked().into(); @@ -369,77 +500,9 @@ impl + Send> SegmentCounts { } } -impl< - NS: NodeSegmentOps, - ES: EdgeSegmentOps, - GS: GraphPropSegmentOps, - EXT: PersistenceStrategy, -> Drop for GraphStore -{ - 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(); - - // Skip running a clean flush if the DB is in shutdown or crash recovery. - // Note that the state can be Shutdown after load is called and before recovery is complete. - // A clean flush is performed even when state is WalDisabled or NotSupported. - if matches!( - control_file.db_state(), - DBState::Shutdown | DBState::CrashRecovery - ) { - return; - } - - match self.flush() { - Ok(_) => { - // Log a checkpoint record in the WAL, indicating that the DB was shutdown - // with all the segments flushed to disk. - // On startup, recovery is skipped since there are no pending writes to replay. - let checkpoint_lsn = match wal.log_shutdown_checkpoint() { - Ok(lsn) => lsn, - Err(err) => { - drop_error!("Failed to log shutdown checkpoint in drop: {err}"); - // this is unreachable with panic-on-drop - #[allow(unreachable_code)] - return; - } - }; - - // Flush up to the end of the WAL stream. - let flush_lsn = wal.position(); - - if let Err(err) = wal.flush(flush_lsn) { - drop_error!("Failed to flush checkpoint record in drop: {err}"); - // this is unreachable with panic-on-drop - #[allow(unreachable_code)] - return; - } - - // Record the checkpoint and shutdown state and write control file to disk. - control_file.set_checkpoint(checkpoint_lsn); - control_file.set_db_state(DBState::Shutdown); - - if let Err(err) = control_file.save() { - drop_error!("Failed to save control file in drop: {err}"); - // this is unreachable with panic-on-drop - #[allow(unreachable_code)] - return; - } - } - Err(err) => { - drop_error!("Failed to flush storage in drop: {err}"); - } - } - } -} +// `GraphStore` has no explicit `Drop` — the `shutdown_guard` field's +// Drop runs the clean-shutdown protocol on writable handles, and is +// simply absent (`None`) on read-only handles. #[inline(always)] pub fn resolve_pos>(i: I, max_page_len: u32) -> (usize, LocalPOS) { diff --git a/raphtory/src/db/api/storage/storage.rs b/raphtory/src/db/api/storage/storage.rs index bc2e854b3e..ffba35ddb3 100644 --- a/raphtory/src/db/api/storage/storage.rs +++ b/raphtory/src/db/api/storage/storage.rs @@ -62,31 +62,47 @@ 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, + // `Some` on writable handles; on drop the guard refreshes the + // on-disk `.meta` counts. `None` on read-only handles, whose drop + // is therefore a no-op with respect to the graph directory. + #[cfg(feature = "io")] + metadata_guard: Option, #[cfg(feature = "search")] pub(crate) index: RwLock, } +/// Refreshes the disk-backed `.meta` file on drop. Held only by +/// writable `Storage` handles. Reader handles omit this guard, so a +/// read-only lifecycle can never touch `.meta`. #[cfg(feature = "io")] -impl Drop for Storage { +#[derive(Debug)] +struct MetadataRefreshGuard { + disk_path: std::path::PathBuf, + graph: GraphStorage, +} + +#[cfg(feature = "io")] +impl Drop for MetadataRefreshGuard { 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); - let edge_count = self.graph.unfiltered_num_edges(&LayerIds::All); - // Drop must not panic - ignore any error refreshing the metadata - // file. The graph data itself is already persisted by the storage - // layer so a stale `.meta` only affects node and edge counts (for now). - let _ = storage::refresh_disk_graph_metadata(&disk_path, node_count, edge_count); - } + let node_count = self.graph.unfiltered_num_nodes(&LayerIds::All); + let edge_count = self.graph.unfiltered_num_edges(&LayerIds::All); + // Drop must not panic - ignore any error refreshing the metadata + // file. The graph data itself is already persisted by the storage + // layer so a stale `.meta` only affects node and edge counts (for now). + let _ = storage::refresh_disk_graph_metadata(&self.disk_path, node_count, edge_count); + } +} + +// `Storage` has no explicit `Drop` — the `metadata_guard` field's Drop +// runs the `.meta` refresh on writable handles, and is simply absent +// (`None`) on read-only handles. +#[cfg(feature = "io")] +impl Storage { + fn build_metadata_guard(graph: &GraphStorage) -> Option { + graph.disk_storage_path().map(|p| MetadataRefreshGuard { + disk_path: p.to_path_buf(), + graph: graph.clone(), + }) } } @@ -126,10 +142,12 @@ impl Storage { let config = Config::default(); let ext = Extension::new(config, Some(path.as_ref()))?; let temporal_graph = TemporalGraph::new_at_path_with_ext(path, ext)?; + let graph = GraphStorage::Unlocked(Arc::new(temporal_graph)); Ok(Self { - graph: GraphStorage::Unlocked(Arc::new(temporal_graph)), - read_only: false, + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -138,9 +156,11 @@ impl Storage { pub(crate) fn new_with_config(config: Config) -> Result { let ext = Extension::new(config, None)?; let temporal_graph = TemporalGraph::new(ext)?; + let graph = GraphStorage::Unlocked(Arc::new(temporal_graph)); Ok(Self { - graph: GraphStorage::Unlocked(Arc::new(temporal_graph)), - read_only: false, + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -152,10 +172,12 @@ impl Storage { ) -> Result { let ext = Extension::new(config, Some(path.as_ref()))?; let temporal_graph = TemporalGraph::new_at_path_with_ext(path, ext)?; + let graph = GraphStorage::Unlocked(Arc::new(temporal_graph)); Ok(Self { - graph: GraphStorage::Unlocked(Arc::new(temporal_graph)), - read_only: false, + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -166,10 +188,12 @@ impl Storage { // Run crash recovery if needed. temporal_graph.run_recovery()?; + let graph = GraphStorage::Unlocked(Arc::new(temporal_graph)); Ok(Self { - graph: GraphStorage::Unlocked(Arc::new(temporal_graph)), - read_only: false, + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -187,7 +211,9 @@ impl Storage { Ok(Self { graph: GraphStorage::Mem(locked), - read_only: true, + // Read-only: no metadata guard, drop must not touch `.meta`. + #[cfg(feature = "io")] + metadata_guard: None, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -229,8 +255,9 @@ impl Storage { pub(crate) fn from_inner(graph: GraphStorage) -> Self { Self { + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), graph, - read_only: false, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), } @@ -247,7 +274,8 @@ impl Storage { // 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 = "io")] + metadata_guard: None, #[cfg(feature = "search")] index: RwLock::new(self.index.read().clone()), } From 4d362643db63c0f220c89a433b893b8836faafad Mon Sep 17 00:00:00 2001 From: shivamka1 <4599890+shivamka1@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:42:33 +0100 Subject: [PATCH 4/6] move test file to correct loc --- .../{test_graphql => test_graphdb}/test_read_only_durability.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename python/tests/test_base_install/{test_graphql => test_graphdb}/test_read_only_durability.py (100%) diff --git a/python/tests/test_base_install/test_graphql/test_read_only_durability.py b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py similarity index 100% rename from python/tests/test_base_install/test_graphql/test_read_only_durability.py rename to python/tests/test_base_install/test_graphdb/test_read_only_durability.py From 913629e081bb2ae2861ff41060d61375382399ab Mon Sep 17 00:00:00 2001 From: shivamka1 <4599890+shivamka1@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:46:06 +0100 Subject: [PATCH 5/6] include both event and persistent in the tests --- .../test_graphdb/test_read_only_durability.py | 60 ++++++++++++------- 1 file changed, 39 insertions(+), 21 deletions(-) diff --git a/python/tests/test_base_install/test_graphdb/test_read_only_durability.py b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py index e603d5860c..4b72b4a504 100644 --- a/python/tests/test_base_install/test_graphdb/test_read_only_durability.py +++ b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py @@ -18,6 +18,10 @@ Bugs #1 and #2 are Drop side-effects on shared state. Bug #3 is a reader-open race, independent of Drop. +Each test is parametrised over ``Graph`` and ``PersistentGraph`` — the +underlying storage layer is shared, but the Python entry points are +separate ``#[pymethods]`` blocks that could drift independently. + See ``docs/db-v4-wal-explainer.md`` for the underlying mechanism. """ @@ -38,27 +42,32 @@ ) -# 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 = """ +GRAPH_CLASSES = ["Graph", "PersistentGraph"] + + +def _writer_script(graph_cls_name: str) -> str: + """Build the subprocess writer script for the given graph class.""" + # Double braces in the f-string escape to a literal single brace, so + # ``{{i}}`` becomes ``{i}`` in the emitted script (an f-string in the + # subprocess), and ``{{"v": i}}`` becomes ``{"v": i}`` (a dict). + return f""" import sys, time -from raphtory import Graph -g = Graph(sys.argv[1]) +from raphtory import {graph_cls_name} +g = {graph_cls_name}(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}) + g.add_node(1 + i, f"n{{i}}", properties={{"v": i}}) i += 1 g.flush() time.sleep(0.01) """ -def _spawn_writer(path): +def _spawn_writer(path, graph_cls_name): p = subprocess.Popen( - [sys.executable, "-c", _WRITER_SCRIPT, path], + [sys.executable, "-c", _writer_script(graph_cls_name), path], stdout=subprocess.PIPE, text=True, ) @@ -86,31 +95,39 @@ def _hash_all_files(root): } +def _graph_cls(graph_cls_name): + import raphtory + + return getattr(raphtory, graph_cls_name) + + # --- Test 1: CONTROL -------------------------------------------------------- -def test_writer_crash_recovers_cleanly_without_readers(tmp_path): +@pytest.mark.parametrize("graph_cls_name", GRAPH_CLASSES) +def test_writer_crash_recovers_cleanly_without_readers(tmp_path, graph_cls_name): """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 + graph_cls = _graph_cls(graph_cls_name) path = str(tmp_path / "g") - w = _spawn_writer(path) + w = _spawn_writer(path, graph_cls_name) try: time.sleep(1.5) finally: _kill(w) - g = Graph.load(path) + g = graph_cls.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): +@pytest.mark.parametrize("graph_cls_name", GRAPH_CLASSES) +def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path, graph_cls_name): """Bugs #1 and #2: a read-only ``Graph.load(...)`` followed by drop must not modify any file in the graph directory. @@ -126,10 +143,10 @@ def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path): * Bug #2 - Drop for Storage rewrites ``.meta``, changing its bytes (and racing on ``.tmp`` under concurrency). """ - from raphtory import Graph + graph_cls = _graph_cls(graph_cls_name) path = Path(tmp_path) / "g" - w = _spawn_writer(str(path)) + w = _spawn_writer(str(path), graph_cls_name) # 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) @@ -138,7 +155,7 @@ def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path): try: before = _hash_all_files(path) - rg = Graph.load(str(path), read_only=True) + rg = graph_cls.load(str(path), read_only=True) rg.count_nodes() del rg @@ -162,6 +179,7 @@ def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path): # --- Test 3: Bug #3 (reader-open race, xfail placeholder) ------------------ +@pytest.mark.parametrize("graph_cls_name", GRAPH_CLASSES) @pytest.mark.xfail( reason=( "Bug #3: reader Graph.load races the writer's segment-file creation; " @@ -171,7 +189,7 @@ def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path): ), strict=False, ) -def test_reader_open_does_not_race_writer_segment_creation(tmp_path): +def test_reader_open_does_not_race_writer_segment_creation(tmp_path, graph_cls_name): """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. @@ -179,10 +197,10 @@ def test_reader_open_does_not_race_writer_segment_creation(tmp_path): 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 + graph_cls = _graph_cls(graph_cls_name) path = str(tmp_path / "g") - w = _spawn_writer(path) + w = _spawn_writer(path, graph_cls_name) stop = threading.Event() enoent_errors = [] @@ -191,7 +209,7 @@ def test_reader_open_does_not_race_writer_segment_creation(tmp_path): def reader(): while not stop.is_set(): try: - rg = Graph.load(path, read_only=True) + rg = graph_cls.load(path, read_only=True) rg.count_nodes() del rg except Exception as exc: # noqa: BLE001 - collect any error From 6e3c10abdfee04970a52c90e9a1a2656896a847b Mon Sep 17 00:00:00 2001 From: shivamka1 <4599890+shivamka1@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:55:54 +0100 Subject: [PATCH 6/6] mov storage test to storage --- .../test_graphdb/test_read_only_durability.py | 104 ++++-------------- 1 file changed, 21 insertions(+), 83 deletions(-) diff --git a/python/tests/test_base_install/test_graphdb/test_read_only_durability.py b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py index 4b72b4a504..c1ea3f5434 100644 --- a/python/tests/test_base_install/test_graphdb/test_read_only_durability.py +++ b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py @@ -1,28 +1,28 @@ """ -Durability regression tests for concurrent read-only opens. +Durability regression tests for read-only opens. -There are three distinct bugs in this area, targeted by three separate -tests below: +These tests exercise Raphtory's Drop-side-effect fixes for concurrent +read-only opens of a disk-backed graph: - 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 + Bug #1 Reader Drop appended 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 failed 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``. + Bug #2 Reader Drop rewrote ``.meta`` via a fixed-name ``.tmp`` file. + Racing renames intermittently ENOENT'd at + ``meta_file.rs:80``. -Bugs #1 and #2 are Drop side-effects on shared state. Bug #3 is a -reader-open race, independent of Drop. +Both bugs are fixed by Raphtory-side Drop guards (see +``WriterShutdownGuard`` on ``GraphStore`` in ``db4-storage`` and +``MetadataRefreshGuard`` on ``Storage`` in ``raphtory``). + +A separate reader-vs-writer race in pometry-storage's segment +publication has its own regression test in pometry-storage's +``python/tests/test_read_only_durability.py``. Each test is parametrised over ``Graph`` and ``PersistentGraph`` — the underlying storage layer is shared, but the Python entry points are separate ``#[pymethods]`` blocks that could drift independently. - -See ``docs/db-v4-wal-explainer.md`` for the underlying mechanism. """ import hashlib @@ -30,7 +30,6 @@ import signal import subprocess import sys -import threading import time from pathlib import Path @@ -47,9 +46,10 @@ def _writer_script(graph_cls_name: str) -> str: """Build the subprocess writer script for the given graph class.""" - # Double braces in the f-string escape to a literal single brace, so - # ``{{i}}`` becomes ``{i}`` in the emitted script (an f-string in the - # subprocess), and ``{{"v": i}}`` becomes ``{"v": i}`` (a dict). + # Double braces in the outer f-string escape to a literal single + # brace, so ``{{i}}`` becomes ``{i}`` in the emitted script (an + # f-string in the subprocess), and ``{{"v": i}}`` becomes + # ``{"v": i}`` (a dict). return f""" import sys, time from raphtory import {graph_cls_name} @@ -109,7 +109,7 @@ def test_writer_crash_recovers_cleanly_without_readers(tmp_path, graph_cls_name) """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.""" + case below.""" graph_cls = _graph_cls(graph_cls_name) path = str(tmp_path / "g") @@ -174,65 +174,3 @@ def test_readonly_open_and_drop_does_not_modify_any_file(tmp_path, graph_cls_nam 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.parametrize("graph_cls_name", GRAPH_CLASSES) -@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, graph_cls_name): - """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. - """ - graph_cls = _graph_cls(graph_cls_name) - - path = str(tmp_path / "g") - w = _spawn_writer(path, graph_cls_name) - - stop = threading.Event() - enoent_errors = [] - lock = threading.Lock() - - def reader(): - while not stop.is_set(): - try: - rg = graph_cls.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]}" - )