Implement wal logging and replay for EdgeView, NodeView, delete_edge and graph properties

db4-storage/src/api/graph_props.rs

@@ -1,4 +1,4 @@
-use crate::{error::StorageError, segments::graph_prop::segment::MemGraphPropSegment};
+use crate::{error::StorageError, segments::graph_prop::segment::MemGraphPropSegment, wal::LSN};
 use parking_lot::{RwLockReadGuard, RwLockWriteGuard};
 use raphtory_api::core::entities::properties::{meta::Meta, prop::Prop, tprop::TPropOps};
 use std::{fmt::Debug, path::Path, sync::Arc};
@@ -31,6 +31,9 @@ where
 
     fn set_dirty(&self, dirty: bool);
 
+    /// Returns the latest lsn for the immutable part of this segment.
+    fn immut_lsn(&self) -> LSN;
+
     fn notify_write(
         &self,
         mem_segment: &mut RwLockWriteGuard<'_, MemGraphPropSegment>,

db4-storage/src/pages/edge_page/writer.rs

@@ -1,6 +1,6 @@
 use crate::{
     LocalPOS, api::edges::EdgeSegmentOps, error::StorageError, pages::layer_counter::GraphStats,
-    segments::edge::segment::MemEdgeSegment,
+    segments::edge::segment::MemEdgeSegment, wal::LSN,
 };
 use raphtory_api::core::entities::{
     VID,
@@ -72,9 +72,11 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
         let existing_edge = self
             .page
             .contains_edge(edge_pos, layer_id, self.writer.deref());
+
         if !existing_edge {
             self.increment_layer_num_edges(layer_id);
         }
+
         self.graph_stats.update_time(t.t());
         self.writer
             .delete_edge_internal(t, edge_pos, src, dst, layer_id);
@@ -166,6 +168,10 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
         self.page.get_edge(edge_pos, layer_id, self.writer.deref())
     }
 
+    pub fn set_lsn(&mut self, lsn: LSN) {
+        self.writer.set_lsn(lsn);
+    }
+
     pub fn check_metadata(
         &self,
         edge_pos: LocalPOS,

db4-storage/src/pages/graph_prop_page/writer.rs

@@ -1,6 +1,6 @@
 use crate::{
     api::graph_props::GraphPropSegmentOps, error::StorageError,
-    segments::graph_prop::segment::MemGraphPropSegment,
+    segments::graph_prop::segment::MemGraphPropSegment, wal::LSN,
 };
 use parking_lot::RwLockWriteGuard;
 use raphtory_api::core::entities::properties::prop::Prop;
@@ -35,16 +35,20 @@ impl<'a, GS: GraphPropSegmentOps> GraphPropWriter<'a, GS> {
         self.graph_props.set_dirty(true);
     }
 
-    pub fn check_metadata(&self, props: &[(usize, Prop)]) -> Result<(), StorageError> {
-        self.mem_segment.check_metadata(props)
-    }
-
     pub fn update_metadata(&mut self, props: impl IntoIterator<Item = (usize, Prop)>) {
         let add = self.mem_segment.update_metadata(props);
 
         self.graph_props.increment_est_size(add);
         self.graph_props.set_dirty(true);
     }
+
+    pub fn check_metadata(&self, props: &[(usize, Prop)]) -> Result<(), StorageError> {
+        self.mem_segment.check_metadata(props)
+    }
+
+    pub fn set_lsn(&mut self, lsn: LSN) {
+        self.mem_segment.set_lsn(lsn);
+    }
 }
 
 impl<GS: GraphPropSegmentOps> Drop for GraphPropWriter<'_, GS> {

db4-storage/src/pages/locked/graph_props.rs

@@ -1,5 +1,6 @@
 use crate::{
     api::graph_props::GraphPropSegmentOps, segments::graph_prop::segment::MemGraphPropSegment,
+    wal::LSN,
 };
 use parking_lot::RwLockWriteGuard;
 use raphtory_api::core::entities::properties::prop::Prop;
@@ -43,6 +44,10 @@ impl<'a, GS: GraphPropSegmentOps> LockedGraphPropPage<'a, GS> {
         self.page.increment_est_size(add);
         self.page.set_dirty(true);
     }
+
+    pub fn set_lsn(&mut self, lsn: LSN) {
+        self.lock.set_lsn(lsn);
+    }
 }
 
 impl<GS: GraphPropSegmentOps> Drop for LockedGraphPropPage<'_, GS> {
@@ -54,23 +59,15 @@ impl<GS: GraphPropSegmentOps> Drop for LockedGraphPropPage<'_, GS> {
 }
 
 pub struct WriteLockedGraphPropPages<'a, GS: GraphPropSegmentOps> {
-    writer: Option<LockedGraphPropPage<'a, GS>>,
-}
-
-impl<GS: GraphPropSegmentOps> Default for WriteLockedGraphPropPages<'_, GS> {
-    fn default() -> Self {
-        Self { writer: None }
-    }
+    writer: LockedGraphPropPage<'a, GS>,
 }
 
 impl<'a, GS: GraphPropSegmentOps> WriteLockedGraphPropPages<'a, GS> {
     pub fn new(writer: LockedGraphPropPage<'a, GS>) -> Self {
-        Self {
-            writer: Some(writer),
-        }
+        Self { writer }
     }
 
-    pub fn writer(&mut self) -> Option<&mut LockedGraphPropPage<'a, GS>> {
-        self.writer.as_mut()
+    pub fn writer(&mut self) -> &mut LockedGraphPropPage<'a, GS> {
+        &mut self.writer
     }
 }

db4-storage/src/pages/mod.rs

@@ -77,12 +77,15 @@ impl<
     pub 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);
+
         if let Some(graph_dir) = self.graph_dir.as_ref() {
             config.save_to_dir(graph_dir)?;
         }
+
         self.nodes.flush()?;
         self.edges.flush()?;
         self.graph_props.flush()?;
+
         Ok(())
     }
 }

db4-storage/src/pages/node_page/writer.rs

@@ -1,6 +1,6 @@
 use crate::{
     LocalPOS, api::nodes::NodeSegmentOps, error::StorageError, pages::layer_counter::GraphStats,
-    segments::node::segment::MemNodeSegment,
+    segments::node::segment::MemNodeSegment, wal::LSN,
 };
 use raphtory_api::core::entities::{
     EID, GID, VID,
@@ -222,6 +222,10 @@ impl<'a, MP: DerefMut<Target = MemNodeSegment> + 'a, NS: NodeSegmentOps> NodeWri
     pub fn has_node(&self, node: LocalPOS, layer_id: usize) -> bool {
         self.mut_segment.has_node(node, layer_id) || self.page.has_node(node, layer_id)
     }
+
+    pub fn set_lsn(&mut self, lsn: LSN) {
+        self.mut_segment.set_lsn(lsn);
+    }
 }
 
 pub fn node_info_as_props(
@@ -260,4 +264,11 @@ impl<'a, MP: DerefMut<Target = MemNodeSegment>, NS: NodeSegmentOps> NodeWriters<
     pub fn get_mut_dst(&mut self) -> &mut NodeWriter<'a, MP, NS> {
         self.dst.as_mut().unwrap_or(&mut self.src)
     }
+
+    pub fn set_lsn(&mut self, lsn: LSN) {
+        self.src.set_lsn(lsn);
+        if let Some(dst) = &mut self.dst {
+            dst.set_lsn(lsn);
+        }
+    }
 }

db4-storage/src/pages/node_store.rs

@@ -33,7 +33,10 @@ pub const N: usize = 32;
 pub struct NodeStorageInner<NS, EXT> {
     segments: boxcar::Vec<Arc<NS>>,
     stats: Arc<GraphStats>,
+
+    /// Contains ids of segments that can accomodate new nodes.
     free_segments: Box<[RwLock<usize>; N]>,
+
     nodes_path: Option<PathBuf>,
     node_meta: Arc<Meta>,
     edge_meta: Arc<Meta>,
@@ -281,59 +284,47 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy> NodeStorageI
         }
     }
 
-    /// Reserves space in a segment and returns a locked writer for that segment.
-    ///
-    /// This method implements a lock-free reservation strategy with lazy segment allocation:
-    ///
-    /// **What it does:**
-    /// 1. Uses the `row` parameter to determine a starting slot in the `free_segments` array (via modulo N)
-    /// 2. Attempts to acquire a write lock on the segment at that slot
-    /// 3. If successful, tries to atomically reserve `required_space` rows in that segment
-    /// 4. If the segment is full or locked, tries the next slot (round-robin fashion)
-    /// 5. When a segment becomes full, creates a new segment and updates the slot atomically
-    /// 6. Returns both the reserved position and a locked writer for immediate use
-    ///
-    /// **Why it works this way:**
-    /// - **Lock-free fast path**: Uses `try_writer` to avoid blocking on contended segments
-    /// - **Load distribution**: The `row % N` ensures different threads naturally distribute across slots
-    /// - **Atomic reservation**: `reserve_segment_rows` uses atomic compare-and-swap to reserve space
-    ///   without holding locks during the reservation attempt
-    /// - **Double-check locking**: When creating a new segment, re-checks if another thread already
-    ///   created one (via `if *slot == page_id`) to avoid unnecessary allocations
-    /// - **Keeps writer locked**: Returns the writer still locked so the caller can immediately write
-    ///   to the reserved space without risk of concurrent modification
-    /// - **Circular probing**: If a slot is contended or full, moves to the next slot rather than
-    ///   blocking, maximizing throughput in multi-threaded scenarios
-    ///
-    /// This design prioritizes:
-    /// - High concurrency: Multiple threads can reserve space simultaneously in different segments
-    /// - Low latency: Avoids blocking whenever possible by trying different segments
-    /// - Memory efficiency: Only creates new segments when actually needed
+    /// Select a segment using `row` as a hint and reserves `num_rows` in that segment.
+    /// Returns the reserved position and a locked writer for that segment.
     pub fn reserve_and_lock_segment(
         &self,
         row: usize,
-        required_space: u32,
+        num_rows: u32,
     ) -> (
         LocalPOS,
         NodeWriter<'_, RwLockWriteGuard<'_, MemNodeSegment>, NS>,
     ) {
         let mut slot_idx = row % N;
-        let mut page_id = *self.free_segments[slot_idx].read_recursive();
+        let mut segment_id = *self.free_segments[slot_idx].read_recursive();
+
+        // Iterate through `free_segments` until we can acquire a write lock on a free segment.
+        // With the write lock, try reserving `num_rows` in that segment.
+        // If unsuccessful, find another free segment and try again, creating a new one if needed.
         loop {
-            match self.try_writer(page_id) {
+            match self.try_writer(segment_id) {
                 None => {
+                    // The current segment is being written to, round-robin to the next slot.
                     slot_idx = (slot_idx + 1) % N;
-                    page_id = *self.free_segments[slot_idx].read_recursive();
+                    let slot = self.free_segments[slot_idx].read_recursive();
+                    segment_id = *slot;
                 }
                 Some(writer) => {
-                    match self.reserve_segment_rows(writer.page, required_space) {
+                    match self.reserve_segment_rows(writer.page, num_rows) {
                         None => {
-                            // segment is full, we need to create a new one
+                            // The current segment is full, drop its lock and try to find
+                            // another free segment.
+                            drop(writer);
+
                             let mut slot = self.free_segments[slot_idx].write();
-                            if *slot == page_id {
-                                // page_id is unchanged, no other thread created a new segment before we got the lock
-                                page_id = self.push_new_segment();
-                                *slot = page_id;
+
+                            // Check our slot to see if some other thread has pushed a new free segment.
+                            if *slot == segment_id {
+                                // The segment is still the same, so we need to create a new one.
+                                segment_id = self.push_new_segment();
+                                *slot = segment_id;
+                            } else {
+                                // There is another free segment in the slot, retry using it.
+                                segment_id = *slot;
                             }
                         }
                         Some(local_pos) => return (LocalPOS(local_pos), writer),
@@ -344,10 +335,13 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy> NodeStorageI
     }
 
     /// Reserves a single row in the given segment and returns the position if successful.
+    /// Returns `None` if the segment is full.
     pub fn reserve_segment_row(&self, segment: &NS) -> Option<u32> {
         self.reserve_segment_rows(segment, 1)
     }
 
+    /// Reserves `rows` in the given segment and returns the position if successful.
+    /// Returns `None` if the segment is full.
     fn reserve_segment_rows(&self, segment: &NS, rows: u32) -> Option<u32> {
         increment_and_clamp(segment.nodes_counter(), rows, self.max_segment_len())
     }
@@ -637,25 +631,17 @@ impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy> NodeStorageI
     }
 }
 
-/// Atomically increments a counter and returns the previous value, but only if the result stays within bounds.
-///
-/// 1. Atomically reads the current counter value
-/// 2. Computes `current + increment`
-/// 3. If the result is ≤ `max_segment_len`, updates the counter and returns the *previous* value
-/// 4. If the result would exceed the limit, leaves the counter unchanged and returns `None`
-///
-pub fn increment_and_clamp(
-    counter: &AtomicU32,
-    increment: u32,
-    max_segment_len: u32,
-) -> Option<u32> {
+/// Atomically increments `counter` and returns the previous value, but only if the result stays
+/// within bounds.
+/// If the result exceeds `limit`, leaves the counter unchanged and returns `None`.
+pub fn increment_and_clamp(counter: &AtomicU32, increment: u32, limit: u32) -> Option<u32> {
     counter
         .fetch_update(
             std::sync::atomic::Ordering::Relaxed,
             std::sync::atomic::Ordering::Relaxed,
             |current| {
                 let updated = current + increment;
-                if updated <= max_segment_len {
+                if updated <= limit {
                     Some(updated)
                 } else {
                     None

db4-storage/src/pages/session.rs

@@ -9,14 +9,11 @@ use crate::{
 };
 use parking_lot::RwLockWriteGuard;
 use raphtory_api::core::{
-    entities::properties::{
-        meta::{NODE_ID_IDX, STATIC_GRAPH_LAYER_ID},
-        prop::Prop,
-    },
+    entities::properties::{meta::STATIC_GRAPH_LAYER_ID, prop::Prop},
     storage::dict_mapper::MaybeNew,
 };
 use raphtory_core::{
-    entities::{EID, ELID, GidRef, VID},
+    entities::{EID, ELID, VID},
     storage::timeindex::AsTime,
 };
 
@@ -52,30 +49,6 @@ impl<
         }
     }
 
-    pub fn store_src_node_info(&mut self, vid: impl Into<VID>, node_id: Option<GidRef>) {
-        if let Some(id) = node_id {
-            let pos = self.graph.nodes().resolve_pos(vid.into()).1;
-
-            self.node_writers().get_mut_src().update_c_props(
-                pos,
-                STATIC_GRAPH_LAYER_ID,
-                [(NODE_ID_IDX, id.into())],
-            );
-        }
-    }
-
-    pub fn store_dst_node_info(&mut self, vid: impl Into<VID>, node_id: Option<GidRef>) {
-        if let Some(id) = node_id {
-            let pos = self.graph.nodes().resolve_pos(vid.into()).1;
-
-            self.node_writers().get_mut_dst().update_c_props(
-                pos,
-                STATIC_GRAPH_LAYER_ID,
-                [(NODE_ID_IDX, id.into())],
-            );
-        }
-    }
-
     pub fn add_edge_into_layer<T: AsTime>(
         &mut self,
         t: T,
@@ -168,6 +141,7 @@ impl<
                 self.node_writers
                     .get_mut_src()
                     .add_outbound_edge(Some(t), src_pos, dst, edge_id);
+
                 self.node_writers
                     .get_mut_dst()
                     .add_inbound_edge(Some(t), dst_pos, src, edge_id);
@@ -176,6 +150,7 @@ impl<
             self.node_writers
                 .get_mut_src()
                 .update_deletion_time(t, src_pos, e_id);
+
             self.node_writers
                 .get_mut_dst()
                 .update_deletion_time(t, dst_pos, e_id);
@@ -226,12 +201,7 @@ impl<
     }
 
     pub fn set_lsn(&mut self, lsn: LSN) {
-        self.node_writers.src.mut_segment.set_lsn(lsn);
-
-        if let Some(dst) = &mut self.node_writers.dst {
-            dst.mut_segment.set_lsn(lsn);
-        }
-
-        self.edge_writer.writer.set_lsn(lsn);
+        self.node_writers.set_lsn(lsn);
+        self.edge_writer.set_lsn(lsn);
     }
 }