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..329150f0cb 100644 --- a/db4-storage/src/pages/mod.rs +++ b/db4-storage/src/pages/mod.rs @@ -57,6 +57,32 @@ pub struct GraphStore< graph_dir: Option, event_id: AtomicUsize, ext: EXT, + // `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< @@ -64,9 +90,9 @@ impl< ES: EdgeSegmentOps, GS: GraphPropSegmentOps, EXT: PersistenceStrategy, -> GraphStore +> WriterShutdownGuard { - pub fn flush(&self) -> Result<(), StorageError> { + 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); @@ -81,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, @@ -146,17 +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, + 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"); @@ -184,13 +315,30 @@ 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, + shutdown_guard, }) } @@ -352,69 +500,9 @@ impl + Send> SegmentCounts { } } -impl< - NS: NodeSegmentOps, - ES: EdgeSegmentOps, - GS: GraphPropSegmentOps, - EXT: PersistenceStrategy, -> Drop for GraphStore -{ - 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}"); - } - } - } -} +// `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/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 new file mode 100644 index 0000000000..c1ea3f5434 --- /dev/null +++ b/python/tests/test_base_install/test_graphdb/test_read_only_durability.py @@ -0,0 +1,176 @@ +""" +Durability regression tests for read-only opens. + +These tests exercise Raphtory's Drop-side-effect fixes for concurrent +read-only opens of a disk-backed graph: + + 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 rewrote ``.meta`` via a fixed-name ``.tmp`` file. + Racing renames intermittently ENOENT'd at + ``meta_file.rs:80``. + +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. +""" + +import hashlib +import os +import signal +import subprocess +import sys +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", +) + + +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 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} +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}}) + i += 1 + g.flush() + time.sleep(0.01) +""" + + +def _spawn_writer(path, graph_cls_name): + p = subprocess.Popen( + [sys.executable, "-c", _writer_script(graph_cls_name), 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() + } + + +def _graph_cls(graph_cls_name): + import raphtory + + return getattr(raphtory, graph_cls_name) + + +# --- Test 1: CONTROL -------------------------------------------------------- + + +@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 + case below.""" + graph_cls = _graph_cls(graph_cls_name) + + path = str(tmp_path / "g") + w = _spawn_writer(path, graph_cls_name) + try: + time.sleep(1.5) + finally: + _kill(w) + + g = graph_cls.load(path) + assert g.count_nodes() > 0 + + +# --- Test 2: Bugs #1 and #2 (deterministic Drop side-effect test) ---------- + + +@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. + + 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). + """ + graph_cls = _graph_cls(graph_cls_name) + + path = Path(tmp_path) / "g" + 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) + + os.kill(w.pid, signal.SIGSTOP) + try: + before = _hash_all_files(path) + + rg = graph_cls.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())}" + ) diff --git a/raphtory/src/db/api/storage/storage.rs b/raphtory/src/db/api/storage/storage.rs index bc5e8b7862..ffba35ddb3 100644 --- a/raphtory/src/db/api/storage/storage.rs +++ b/raphtory/src/db/api/storage/storage.rs @@ -62,22 +62,47 @@ pub use storage::{ #[derive(Debug, Default)] pub struct Storage { graph: GraphStorage, + // `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) { - 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(), + }) } } @@ -117,9 +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)), + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -128,8 +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)), + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -141,9 +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)), + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -154,9 +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)), + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), + graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -174,6 +211,9 @@ impl Storage { Ok(Self { graph: GraphStorage::Mem(locked), + // Read-only: no metadata guard, drop must not touch `.meta`. + #[cfg(feature = "io")] + metadata_guard: None, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), }) @@ -215,6 +255,8 @@ impl Storage { pub(crate) fn from_inner(graph: GraphStorage) -> Self { Self { + #[cfg(feature = "io")] + metadata_guard: Self::build_metadata_guard(&graph), graph, #[cfg(feature = "search")] index: RwLock::new(GraphIndex::Empty), @@ -229,6 +271,11 @@ 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. + #[cfg(feature = "io")] + metadata_guard: None, #[cfg(feature = "search")] index: RwLock::new(self.index.read().clone()), }