From 89c74ff0534f303d98eab5100200d56a423d7c39 Mon Sep 17 00:00:00 2001 From: Yu-da-1 Date: Sun, 9 Nov 2025 23:00:40 +0900 Subject: [PATCH 1/2] fix: upate database --- examples/cli.rs | 13 +- examples/content_versioning.rs | 6 +- readme.md | 8 +- src/crdt/crdt_state.rs | 14 + src/crdt/storage.rs | 267 ++++++++-------- src/graph/dag.rs | 269 +++++++++++----- src/graph/storage.rs | 110 ++++--- src/lib.rs | 1 + src/repo.rs | 551 +++++++++++++++++++++++---------- src/storage/mod.rs | 3 + src/storage/shared_leveldb.rs | 160 ++++++++++ 11 files changed, 968 insertions(+), 434 deletions(-) create mode 100644 src/storage/mod.rs create mode 100644 src/storage/shared_leveldb.rs diff --git a/examples/cli.rs b/examples/cli.rs index f36bebf..e9f2a3d 100644 --- a/examples/cli.rs +++ b/examples/cli.rs @@ -9,6 +9,7 @@ use crsl_lib::crdt::{ use crsl_lib::dasl::cid::ContentId; use crsl_lib::graph::{dag::DagGraph, storage::LeveldbNodeStorage}; use crsl_lib::repo::Repo; +use crsl_lib::storage::SharedLeveldb; use std::collections::{HashMap, HashSet}; use std::error::Error; use std::path::{Path, PathBuf}; @@ -63,12 +64,11 @@ fn main() -> Result<(), Box> { match cli.cmd { Commands::Init { path } => { std::fs::create_dir_all(&path)?; - std::fs::create_dir_all(path.join("ops"))?; - std::fs::create_dir_all(path.join("nodes"))?; + std::fs::create_dir_all(path.join("store"))?; std::fs::write(path.join(".crsl"), "")?; - println!("Initialized CRSL repository at {path:?}"); + println!("Initialized CRSL repository at {path:?} (single LevelDB store)"); } other_command => { let repo_path = Path::new(DEFAULT_REPO_PATH); @@ -215,10 +215,9 @@ fn main() -> Result<(), Box> { } fn open_repo(repo_path: &Path) -> Result> { - let op_storage = LeveldbStorage::open(repo_path.join("ops"))?; - let node_storage = LeveldbNodeStorage::open(repo_path.join("nodes")); - let state = CrdtState::new(op_storage); - let dag = DagGraph::new(node_storage); + let shared = SharedLeveldb::open(repo_path.join("store"))?; + let state = CrdtState::new(LeveldbStorage::new(shared.clone())); + let dag = DagGraph::new(LeveldbNodeStorage::new(shared)); Ok(Repo::new(state, dag)) } diff --git a/examples/content_versioning.rs b/examples/content_versioning.rs index 22c37f0..c245246 100644 --- a/examples/content_versioning.rs +++ b/examples/content_versioning.rs @@ -14,6 +14,7 @@ use crsl_lib::{ storage::LeveldbStorage as OpStore, }, graph::{dag::DagGraph, storage::LeveldbNodeStorage as NodeStorage}, + storage::SharedLeveldb, }; use tempfile::tempdir; @@ -23,8 +24,9 @@ type ContentState = CrdtState; fn main() { let tmp = tempdir().expect("tmp dir"); - let op_store = OpStore::open(tmp.path().join("ops")).unwrap(); - let node_store = NodeStorage::open(tmp.path().join("nodes")); + let shared = SharedLeveldb::open(tmp.path().join("store")).unwrap(); + let op_store = OpStore::new(shared.clone()); + let node_store = NodeStorage::new(shared); let state = ContentState::new(op_store); let mut _dag = DagGraph::<_, Content, ()>::new(node_store); diff --git a/readme.md b/readme.md index 9a4b45b..5640caa 100644 --- a/readme.md +++ b/readme.md @@ -24,6 +24,7 @@ use crsl_lib::{ }, graph::{dag::DagGraph, storage::LeveldbNodeStorage as NodeStorage}, repo::Repo, + storage::SharedLeveldb, }; use tempfile::tempdir; use cid::Cid; @@ -34,10 +35,9 @@ struct Content(String); fn main() { // Initialize storage let tmp = tempdir().expect("tmp dir"); - let op_store = OpStore::open(tmp.path().join("ops")).unwrap(); - let node_store = NodeStorage::open(tmp.path().join("nodes")); - let state = CrdtState::new(op_store); - let dag = DagGraph::new(node_store); + let shared = SharedLeveldb::open(tmp.path().join("store")).unwrap(); + let state = CrdtState::new(OpStore::new(shared.clone())); + let dag = DagGraph::new(NodeStorage::new(shared)); let mut repo = Repo::new(state, dag); // Create a content ID (in practice, you'd use a proper CID) diff --git a/src/crdt/crdt_state.rs b/src/crdt/crdt_state.rs index 1aa1c3f..dab1708 100644 --- a/src/crdt/crdt_state.rs +++ b/src/crdt/crdt_state.rs @@ -4,6 +4,7 @@ use crate::crdt::reducer::Reducer; use crate::crdt::storage::OperationStorage; use std::fmt::Debug; use std::marker::PhantomData; +use ulid::Ulid; /// A generic CRDT state container that manages operations on content. /// /// `CrdtState` provides a high-level interface for applying operations to content @@ -39,6 +40,10 @@ where _marker: PhantomData, } } + + pub fn storage(&self) -> &S { + &self.storage + } /// Applies an operation to the CRDT state without validation. /// /// This method directly saves the operation to storage without checking its validity. @@ -84,6 +89,14 @@ where self.storage.load_operations(genesis) } + pub fn get_operation(&self, op_id: &Ulid) -> Result>> { + self.storage.get_operation(op_id) + } + + pub fn delete_operation(&self, op_id: &Ulid) -> Result<()> { + self.storage.delete_operation(op_id) + } + /// Validates whether an operation is logically valid to apply. /// /// This method performs the following checks: @@ -124,6 +137,7 @@ mod tests { #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] struct DummyPayload(String); + /// Helper for constructing operations with deterministic timestamps. fn make_op( id: u64, ts: u64, diff --git a/src/crdt/storage.rs b/src/crdt/storage.rs index 2d2e45c..42c8409 100644 --- a/src/crdt/storage.rs +++ b/src/crdt/storage.rs @@ -1,42 +1,91 @@ use crate::crdt::error::{CrdtError, Result}; use crate::crdt::operation::Operation; +use crate::storage::{BatchError, LeveldbBatchGuard, SharedLeveldb, SharedLeveldbAccess}; use bincode; -use rusty_leveldb::{LdbIterator, Options, DB as Database}; -use std::cell::RefCell; +use rusty_leveldb::LdbIterator; use std::marker::PhantomData; use std::path::Path; +use std::rc::Rc; use ulid::Ulid; +/// Abstraction over the persistent storage used by `CrdtState`. pub trait OperationStorage { fn save_operation(&self, op: &Operation) -> Result<()>; fn load_operations(&self, genesis: &ContentId) -> Result>>; fn get_operation(&self, op_id: &Ulid) -> Result>>; + fn delete_operation(&self, op_id: &Ulid) -> Result<()>; + fn begin_batch(&self) -> std::result::Result, BatchError> { + Err(BatchError::Unsupported) + } } +/// LevelDB-backed implementation of [`OperationStorage`]. +#[derive(Clone)] pub struct LeveldbStorage { - db: RefCell, + shared: Rc, _marker: PhantomData<(ContentId, T)>, } impl LeveldbStorage { pub fn open>(path: P) -> Result { - let opts = Options { - create_if_missing: true, - ..Default::default() - }; - let db = Database::open(path, opts).map_err(CrdtError::Storage)?; - Ok(LeveldbStorage { - db: RefCell::new(db), + let shared = SharedLeveldb::open(path).map_err(CrdtError::Storage)?; + Ok(Self::new(shared)) + } + + pub fn new(shared: Rc) -> Self { + Self { + shared, _marker: PhantomData, - }) + } } + /// Builds the LevelDB key prefix used for operations (`0x01` namespace). fn make_key(id: &Ulid) -> Vec { let mut key = Vec::with_capacity(1 + 16); key.push(0x01); key.extend_from_slice(id.to_bytes().as_ref()); key } + + /// Serialises an operation into the binary format persisted in LevelDB. + fn encode_operation(op: &Operation) -> Result> + where + ContentId: serde::Serialize, + T: serde::Serialize, + { + let value = bincode::serde::encode_to_vec(op, bincode::config::standard())?; + Ok(value) + } + + /// Writes value bytes either to the active batch or directly to the DB. + fn put_bytes(&self, key: &[u8], value: &[u8]) -> Result<()> { + if self + .shared + .with_active_batch(|batch| batch.put(key, value)) + .is_none() + { + self.shared.db().borrow_mut().put(key, value)?; + } + Ok(()) + } + + /// Deletes the given key, respecting an active batch if present. + fn delete_key(&self, key: &[u8]) -> Result<()> { + if self + .shared + .with_active_batch(|batch| batch.delete(key)) + .is_none() + { + self.shared.db().borrow_mut().delete(key)?; + } + Ok(()) + } +} + +impl SharedLeveldbAccess for LeveldbStorage { + fn shared_leveldb(&self) -> Option> { + Some(self.shared.clone()) + } } impl OperationStorage for LeveldbStorage @@ -44,25 +93,28 @@ where ContentId: serde::Serialize + for<'de> serde::Deserialize<'de> + PartialEq + std::fmt::Debug, T: serde::Serialize + for<'de> serde::Deserialize<'de> + std::fmt::Debug, { + fn begin_batch(&self) -> std::result::Result, BatchError> { + self.shared.begin_batch() + } + fn save_operation(&self, op: &Operation) -> Result<()> { let key = Self::make_key(&op.id); - let value = bincode::serde::encode_to_vec(op, bincode::config::standard())?; - self.db.borrow_mut().put(&key, &value)?; - Ok(()) + let value = Self::encode_operation(op)?; + self.put_bytes(&key, &value) } fn load_operations(&self, genesis: &ContentId) -> Result>> { let mut result = Vec::new(); let mut iter = self - .db + .shared + .db() .borrow_mut() .new_iter() .map_err(CrdtError::Storage)?; - // todo: Implement efficient search methods iter.seek_to_first(); + let mut key = Vec::new(); let mut value = Vec::new(); - while iter.valid() { iter.current(&mut key, &mut value); if let Ok((op, _)) = bincode::serde::decode_from_slice::, _>( @@ -81,7 +133,7 @@ where fn get_operation(&self, op_id: &Ulid) -> Result>> { let key = Self::make_key(op_id); - match self.db.borrow_mut().get(&key) { + match self.shared.db().borrow_mut().get(&key) { Some(raw) => { let (op, _) = bincode::serde::decode_from_slice::, _>( &raw, @@ -92,156 +144,93 @@ where None => Ok(None), } } + + fn delete_operation(&self, op_id: &Ulid) -> Result<()> { + let key = Self::make_key(op_id); + self.delete_key(&key) + } } #[cfg(test)] mod tests { use super::*; - use crate::crdt::operation::{Operation, OperationType}; + use crate::crdt::operation::OperationType; + use crate::storage::SharedLeveldb; use serde::{Deserialize, Serialize}; use tempfile::tempdir; #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] - struct DummyContentId(String); + struct DummyContentId(u64); - #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] struct DummyPayload(String); - fn setup_test_storage() -> ( + fn make_op(id: u64, payload: &str) -> Operation { + Operation::new( + DummyContentId(id), + OperationType::Update(DummyPayload(payload.to_string())), + "tester".into(), + ) + } + + fn setup_storage() -> ( LeveldbStorage, tempfile::TempDir, ) { let dir = tempdir().unwrap(); - let storage = LeveldbStorage::open(dir.path()).unwrap(); - (storage, dir) + let shared = SharedLeveldb::open(dir.path()).unwrap(); + (LeveldbStorage::new(shared), dir) } #[test] - fn test_save_operation() { - let (storage, _dir) = setup_test_storage(); - let target = DummyContentId("test".into()); - let payload = DummyPayload("test".into()); - let author = "Alice".to_string(); - let op = Operation::new( - target.clone(), - OperationType::Create(payload.clone()), - author.clone(), - ); - + fn save_and_load_roundtrip() { + let (storage, _dir) = setup_storage(); + let op = make_op(1, "hello"); storage.save_operation(&op).unwrap(); - let retrieved_op = storage.get_operation(&op.id); - assert!(retrieved_op.is_ok()); - assert_eq!(retrieved_op.unwrap(), Some(op)); + let retrieved = storage + .get_operation(&op.id) + .unwrap() + .expect("operation should exist"); + assert_eq!(retrieved, op); + + let all = storage.load_operations(&DummyContentId(1)).unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0], op); } #[test] - fn test_get_operation() { - let (storage, _dir) = setup_test_storage(); - let target = DummyContentId("test".into()); - let payload = DummyPayload("test".into()); - let author = "Alice".to_string(); - let op = Operation::new( - target.clone(), - OperationType::Create(payload.clone()), - author.clone(), - ); + fn delete_operation_removes_entry() { + let (storage, _dir) = setup_storage(); + let op = make_op(7, "bye"); storage.save_operation(&op).unwrap(); - let retrieved_op = storage.get_operation(&op.id); - - assert!(retrieved_op.is_ok()); - assert_eq!(retrieved_op.unwrap(), Some(op)); + storage.delete_operation(&op.id).unwrap(); + assert!(storage.get_operation(&op.id).unwrap().is_none()); } #[test] - fn test_save_and_get_multiple_operations() { - let (storage, _dir) = setup_test_storage(); - let target = DummyContentId("test".into()); - let payload = DummyPayload("test".into()); - let author = "Alice".to_string(); - let op1 = Operation::new( - target.clone(), - OperationType::Create(payload.clone()), - author.clone(), - ); - let op2 = Operation::new( - target.clone(), - OperationType::Update(payload.clone()), - author.clone(), - ); - storage.save_operation(&op1).unwrap(); - storage.save_operation(&op2).unwrap(); - - let retrieved_ops = storage.get_operation(&op1.id); - let retrieved_ops2 = storage.get_operation(&op2.id); - - assert!(retrieved_ops.is_ok()); - assert_eq!(retrieved_ops.unwrap(), Some(op1)); - assert!(retrieved_ops2.is_ok()); - assert_eq!(retrieved_ops2.unwrap(), Some(op2)); - } + fn batch_commit_persists_operations() { + let (storage, _dir) = setup_storage(); - #[test] - fn test_load_operations() { - let (storage, _dir) = setup_test_storage(); - let target = DummyContentId("test".into()); - let genesis = DummyContentId("genesis".into()); - let payload = DummyPayload("test".into()); - let author = "Alice".to_string(); - let op1 = Operation::new( - target.clone(), - OperationType::Create(payload.clone()), - author.clone(), - ); - let op2 = Operation::new( - target.clone(), - OperationType::Update(payload.clone()), - author.clone(), - ); - let op3 = Operation::new( - genesis.clone(), - OperationType::Update(payload.clone()), - author.clone(), - ); - storage.save_operation(&op1).unwrap(); - storage.save_operation(&op2).unwrap(); - storage.save_operation(&op3).unwrap(); - - let retrieved_ops = storage.load_operations(&target); - - assert!(retrieved_ops.is_ok()); - let ops = retrieved_ops.unwrap(); - assert_eq!(ops.len(), 2); - assert!(ops.contains(&op1)); - assert!(ops.contains(&op2)); - } - - /// Demonstrates that an Update with different genesis is **not** returned when querying by target. - #[test] - fn test_same_target_different_genesis_ignored() { - let (storage, _dir) = setup_test_storage(); - let target = DummyContentId("shared".into()); - let payload = DummyPayload("one".into()); - // Create (genesis = target) - let create = Operation::new( - target.clone(), - OperationType::Create(payload.clone()), - "u1".into(), - ); - storage.save_operation(&create).unwrap(); - - // Update with DIFFERENT genesis but same target - let update = Operation::new( - DummyContentId("DIFF".into()), - OperationType::Update(DummyPayload("two".into())), - "u1".into(), - ); - storage.save_operation(&update).unwrap(); - - let ops = storage.load_operations(&target).unwrap(); - // Should contain only the operations belonging to the requested genesis - assert_eq!(ops.len(), 1); - assert!(ops.contains(&create)); + let guard = storage.begin_batch().unwrap(); + let op_a = make_op(10, "a"); + let op_b = make_op(10, "b"); + + storage.save_operation(&op_a).unwrap(); + storage.save_operation(&op_b).unwrap(); + + // Operations are not visible before commit. + assert!(storage.get_operation(&op_a.id).unwrap().is_none()); + + guard.commit().unwrap(); + + assert!(storage.get_operation(&op_a.id).unwrap().is_some()); + assert!(storage.get_operation(&op_b.id).unwrap().is_some()); + + let all = storage.load_operations(&DummyContentId(10)).unwrap(); + assert_eq!(all.len(), 2); + assert!(all.contains(&op_a)); + assert!(all.contains(&op_b)); } } diff --git a/src/graph/dag.rs b/src/graph/dag.rs index 1bb2c89..5ada4a1 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -2,7 +2,7 @@ use crate::dasl::node::Node; use crate::graph::error::{GraphError, Result}; use crate::graph::storage::NodeStorage; use cid::Cid; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::marker::PhantomData; use std::time::{SystemTime, UNIX_EPOCH}; @@ -39,101 +39,137 @@ where } } - /// Add an edge to the graph - /// - /// # Arguments - /// - /// * `payload` - The payload - /// * `parents` - The parent content Ids - /// * `metadata` - The metadata - /// - /// # Returns - /// - /// * `Cid` - The content Id of the new node - /// pub fn add_node(&mut self, payload: P, parents: Vec, metadata: M) -> Result { - let timestamp = Self::current_timestamp()?; - let node = Node::new_genesis(payload, timestamp, metadata); - let new_cid = node.content_id()?; - if self.would_create_cycle_with(&new_cid, &parents)? { - return Err(GraphError::CycleDetected); + if parents.is_empty() { + let (cid, node) = self.prepare_genesis_node(payload, metadata)?; + return self.persist_and_cache(cid, node); } - self.storage.put(&node)?; - - // Update cache incrementally for the new node - self.ensure_subgraph_cached(&parents)?; - for &parent in &parents { - self.edges_forward.entry(parent).or_default().push(new_cid); + let mut inferred_genesis: Option = None; + for parent in &parents { + let node = self + .storage + .get(parent)? + .ok_or(GraphError::NodeNotFound(*parent))?; + let candidate = node.genesis.unwrap_or(*parent); + match inferred_genesis { + Some(existing) if existing != candidate => { + return Err(GraphError::InvalidParent(format!( + "parents belong to different genesis series ({existing:?} vs {candidate:?})" + ))); + } + None => inferred_genesis = Some(candidate), + _ => {} + } } - self.edges_forward.entry(new_cid).or_default(); + let genesis = inferred_genesis.ok_or_else(|| { + GraphError::Internal("child node requires at least one parent".to_string()) + })?; - Ok(new_cid) + let (cid, node) = self.prepare_child_node(payload, parents, genesis, metadata)?; + self.persist_and_cache(cid, node) } - /// Add a genesis node (first version of content) - /// - /// # Arguments - /// - /// * `payload` - The payload - /// * `metadata` - The metadata - /// - /// # Returns - /// - /// * `Cid` - The content Id of the new genesis node - /// pub fn add_genesis_node(&mut self, payload: P, metadata: M) -> Result { + let (cid, node) = self.prepare_genesis_node(payload, metadata)?; + self.persist_and_cache(cid, node) + } + + pub fn add_child_node( + &mut self, + payload: P, + parents: Vec, + genesis: Cid, + metadata: M, + ) -> Result { + let (cid, node) = self.prepare_child_node(payload, parents, genesis, metadata)?; + self.persist_and_cache(cid, node) + } + + pub fn prepare_genesis_node(&mut self, payload: P, metadata: M) -> Result<(Cid, Node)> { let timestamp = Self::current_timestamp()?; let node = Node::new_genesis(payload, timestamp, metadata); let cid = node.content_id()?; - - self.storage.put(&node)?; - - // Initialize cache entry for genesis node - self.edges_forward.entry(cid).or_default(); - - Ok(cid) + Ok((cid, node)) } - /// Add a child node (descendant of an existing node) - /// - /// # Arguments - /// - /// * `payload` - The payload - /// * `parents` - The parent content Ids - /// * `genesis` - The genesis CID that this node belongs to - /// * `metadata` - The metadata - /// - /// # Returns - /// - /// * `Cid` - The content Id of the new child node - /// - pub fn add_child_node( + pub fn prepare_child_node( &mut self, payload: P, parents: Vec, genesis: Cid, metadata: M, - ) -> Result { + ) -> Result<(Cid, Node)> { let timestamp = Self::current_timestamp()?; let node = Node::new_child(payload, parents.clone(), genesis, timestamp, metadata); let cid = node.content_id()?; - // Use optimized genesis-based cycle detection if self.would_create_cycle_with(&cid, &parents)? { return Err(GraphError::CycleDetected); } + Ok((cid, node)) + } + + /// Persists the prepared node and updates the adjacency cache. + fn persist_and_cache(&mut self, cid: Cid, node: Node) -> Result { self.storage.put(&node)?; + self.register_prepared_node(cid, &node)?; + Ok(cid) + } - // Update cache incrementally for the new node - self.ensure_subgraph_cached(&parents)?; - for &parent in &parents { + pub fn register_prepared_node(&mut self, cid: Cid, node: &Node) -> Result<()> { + let parents = node.parents(); + if parents.is_empty() { + self.edges_forward.entry(cid).or_default(); + return Ok(()); + } + + self.ensure_subgraph_cached(parents)?; + for &parent in parents { self.edges_forward.entry(parent).or_default().push(cid); } self.edges_forward.entry(cid).or_default(); + Ok(()) + } - Ok(cid) + pub fn rollback_pending_node(&mut self, cid: &Cid, parents: &[Cid]) { + if let Some(children) = self.edges_forward.get_mut(cid) { + children.clear(); + } + self.edges_forward.remove(cid); + + for parent in parents { + if let Some(children) = self.edges_forward.get_mut(parent) { + children.retain(|child| child != cid); + } + } + } + + pub fn remove_node(&mut self, cid: &Cid) -> Result<()> { + let node = self + .storage + .get(cid)? + .ok_or(GraphError::NodeNotFound(*cid))?; + + if let Some(children) = self.edges_forward.get(cid) { + if !children.is_empty() { + return Err(GraphError::Internal(format!( + "cannot remove node {cid:?} with existing children" + ))); + } + } + + for parent in node.parents() { + if let Some(children) = self.edges_forward.get_mut(parent) { + children.retain(|child| child != cid); + } + } + + self.edges_forward.remove(cid); + self.storage.delete(cid)?; + + Ok(()) } pub fn get_node(&self, cid: &Cid) -> Result>> { @@ -181,17 +217,14 @@ where /// Ensure a subgraph is cached for the given parents and their ancestors /// This implements lazy, incremental cache building fn ensure_subgraph_cached(&mut self, parents: &[Cid]) -> Result<()> { - let mut to_process = Vec::new(); - - // First, check which parents need caching - for &parent in parents { - if !self.edges_forward.contains_key(&parent) { - to_process.push(parent); - } - } + let mut to_process: Vec = parents + .iter() + .copied() + .filter(|parent| !self.edges_forward.contains_key(parent)) + .collect(); // Process nodes that aren't cached yet - let mut processed = std::collections::HashSet::new(); + let mut processed = HashSet::new(); while let Some(current) = to_process.pop() { if processed.contains(¤t) || self.edges_forward.contains_key(¤t) { continue; @@ -223,7 +256,7 @@ where return true; } let mut stack = vec![start]; - let mut visited = std::collections::HashSet::new(); + let mut visited = HashSet::new(); while let Some(node) = stack.pop() { if node == target { return true; @@ -246,7 +279,7 @@ where node_map.insert(*new_cid, parents.to_vec()); let mut to_process = parents.to_vec(); - let mut processed = std::collections::HashSet::new(); + let mut processed = HashSet::new(); while let Some(current_cid) = to_process.pop() { if processed.contains(¤t_cid) { @@ -267,8 +300,8 @@ where pub fn detect_cycle_cid(node_map: &HashMap>) -> Result { let graph = Self::build_adjacency_list(node_map); - let mut visited = std::collections::HashSet::new(); - let mut rec_stack = std::collections::HashSet::new(); + let mut visited = HashSet::new(); + let mut rec_stack = HashSet::new(); for node in graph.keys() { if !visited.contains(node) @@ -302,8 +335,8 @@ where fn has_cycle( node: Cid, graph: &HashMap>, - visited: &mut std::collections::HashSet, - rec_stack: &mut std::collections::HashSet, + visited: &mut HashSet, + rec_stack: &mut HashSet, ) -> bool { visited.insert(node); rec_stack.insert(node); @@ -372,8 +405,8 @@ where } // Returns the set of nodes (CIDs) that are referenced as parents (i.e., nodes that have children) among the given versions. - fn collect_nodes_with_children(&self, nodes: &[Cid]) -> Result> { - let mut has_children = std::collections::HashSet::new(); + fn collect_nodes_with_children(&self, nodes: &[Cid]) -> Result> { + let mut has_children = HashSet::new(); for &node_cid in nodes { if let Some(node) = self.storage.get(&node_cid)? { for parent_cid in node.parents() { @@ -390,7 +423,7 @@ where fn collect_leaf_nodes( &self, nodes: &[Cid], - has_children: &std::collections::HashSet, + has_children: &HashSet, ) -> Result> { let mut leaf_nodes = Vec::new(); for &node_cid in nodes { @@ -407,8 +440,10 @@ where #[cfg(test)] mod tests { use super::*; + use crate::graph::storage::LeveldbNodeStorage; use std::cell::RefCell; use std::collections::BTreeMap; + use tempfile::tempdir; type TestDag = DagGraph>; @@ -945,4 +980,74 @@ mod tests { // unrelated_cid should not be included assert!(!result.contains(&unrelated_cid)); } + + #[test] + fn test_remove_node_without_children() { + let temp_dir = tempdir().unwrap(); + let storage = LeveldbNodeStorage::>::open(temp_dir.path()); + let mut dag = DagGraph::new(storage); + + let genesis = dag + .add_genesis_node("payload".to_string(), BTreeMap::new()) + .unwrap(); + + dag.remove_node(&genesis).unwrap(); + assert!(dag.get_node(&genesis).unwrap().is_none()); + } + + #[test] + fn test_remove_node_with_children_fails() { + let temp_dir = tempdir().unwrap(); + let storage = LeveldbNodeStorage::>::open(temp_dir.path()); + let mut dag = DagGraph::new(storage); + + let genesis = dag + .add_genesis_node("payload".to_string(), BTreeMap::new()) + .unwrap(); + dag.add_child_node("child".to_string(), vec![genesis], genesis, BTreeMap::new()) + .unwrap(); + + let err = dag.remove_node(&genesis); + assert!(err.is_err()); + } + + #[test] + fn test_prepare_register_and_rollback_node() { + let temp_dir = tempdir().unwrap(); + let storage = LeveldbNodeStorage::>::open(temp_dir.path()); + let mut dag = DagGraph::new(storage); + + let (genesis_cid, genesis_node) = dag + .prepare_genesis_node("payload".to_string(), BTreeMap::new()) + .unwrap(); + dag.storage.put(&genesis_node).unwrap(); + dag.register_prepared_node(genesis_cid, &genesis_node) + .unwrap(); + assert!(dag.edges_forward.contains_key(&genesis_cid)); + + let (child_cid, child_node) = dag + .prepare_child_node( + "child".to_string(), + vec![genesis_cid], + genesis_cid, + BTreeMap::new(), + ) + .unwrap(); + + dag.register_prepared_node(child_cid, &child_node).unwrap(); + assert!(dag.edges_forward.contains_key(&child_cid)); + + dag.rollback_pending_node(&child_cid, child_node.parents()); + + assert!( + !dag.edges_forward.contains_key(&child_cid), + "rollback should remove pending child" + ); + if let Some(children) = dag.edges_forward.get(&genesis_cid) { + assert!( + !children.contains(&child_cid), + "rollback should detach child from parent adjacency" + ); + } + } } diff --git a/src/graph/storage.rs b/src/graph/storage.rs index cb13af5..2116919 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -1,12 +1,13 @@ use crate::dasl::node::Node; use crate::graph::error::{GraphError, Result}; +use crate::storage::{SharedLeveldb, SharedLeveldbAccess}; use cid::Cid; -use rusty_leveldb::{LdbIterator, Options, DB as Database}; -use std::cell::RefCell; +use rusty_leveldb::LdbIterator; use std::collections::HashMap; use std::path::Path; -use std::path::PathBuf; +use std::rc::Rc; +/// Minimal interface required for persisting DAG nodes. pub trait NodeStorage { fn get(&self, content_id: &Cid) -> Result>>; fn put(&self, node: &Node) -> Result<()>; @@ -14,46 +15,81 @@ pub trait NodeStorage { fn get_node_map(&self) -> Result>>; } +/// [`NodeStorage`] implementation backed by a shared LevelDB instance. pub struct LeveldbNodeStorage { - db: RefCell, - path: PathBuf, + shared: Rc, _marker: std::marker::PhantomData<(P, M)>, } impl Clone for LeveldbNodeStorage { fn clone(&self) -> Self { - let opts = Options { - create_if_missing: true, - ..Default::default() - }; - let db = Database::open(&self.path, opts).expect("Failed to clone database"); Self { - db: RefCell::new(db), - path: self.path.clone(), + shared: self.shared.clone(), _marker: std::marker::PhantomData, } } } impl LeveldbNodeStorage { + /// Opens LevelDB and wraps it in a shared handle. pub fn open>(path: Pth) -> Self { - let opts = Options { - create_if_missing: true, - ..Default::default() - }; - let db = Database::open(path.as_ref(), opts).unwrap(); + let shared = SharedLeveldb::open(path).expect("Failed to open LevelDB"); + Self::new(shared) + } + + /// Creates the storage from an existing [`SharedLeveldb`] handle. + pub fn new(shared: Rc) -> Self { Self { - db: RefCell::new(db), - path: path.as_ref().to_path_buf(), + shared, _marker: std::marker::PhantomData, } } + + /// Builds the LevelDB key for nodes, prefixed with the `0x10` namespace. fn make_key(cid: &Cid) -> Vec { let mut v = Vec::with_capacity(1 + cid.to_bytes().len()); v.push(0x10); v.extend_from_slice(&cid.to_bytes()); v } + + /// Writes either into the active batch, or directly into the DB if no batch is active. + fn write_bytes(&self, key: &[u8], value: &[u8]) -> Result<()> { + if self + .shared + .with_active_batch(|batch| batch.put(key, value)) + .is_none() + { + self.shared + .db() + .borrow_mut() + .put(key, value) + .map_err(GraphError::Storage)?; + } + Ok(()) + } + + /// Deletes the given key, falling back to the DB when no batch is active. + fn delete_key(&self, key: &[u8]) -> Result<()> { + if self + .shared + .with_active_batch(|batch| batch.delete(key)) + .is_none() + { + self.shared + .db() + .borrow_mut() + .delete(key) + .map_err(GraphError::Storage)?; + } + Ok(()) + } +} + +impl SharedLeveldbAccess for LeveldbNodeStorage { + fn shared_leveldb(&self) -> Option> { + Some(self.shared.clone()) + } } impl NodeStorage for LeveldbNodeStorage @@ -63,7 +99,7 @@ where { fn get(&self, cid: &Cid) -> Result>> { let key = Self::make_key(cid); - match self.db.borrow_mut().get(&key) { + match self.shared.db().borrow_mut().get(&key) { Some(raw) => { let node = Node::from_bytes(&raw).map_err(|e| GraphError::NodeOperation(e.to_string()))?; @@ -81,26 +117,20 @@ where .content_id() .map_err(|e| GraphError::NodeOperation(e.to_string()))?; let key = Self::make_key(&cid); - self.db - .borrow_mut() - .put(&key, &bytes) - .map_err(GraphError::Storage)?; - Ok(()) + self.write_bytes(&key, &bytes) } fn delete(&self, cid: &Cid) -> Result<()> { let key = Self::make_key(cid); - self.db - .borrow_mut() - .delete(&key) - .map_err(GraphError::Storage)?; - Ok(()) + self.delete_key(&key) } + /// Walks all nodes and constructs an adjacency map (parent → children). fn get_node_map(&self) -> Result>> { let mut node_map = HashMap::new(); let mut iter = self - .db + .shared + .db() .borrow_mut() .new_iter() .map_err(GraphError::Storage)?; @@ -111,17 +141,12 @@ where while iter.valid() { iter.current(&mut key, &mut value); if !key.is_empty() && key[0] == 0x10 { - match Node::::from_bytes(&value) { - Ok(node) => { - let node_cid = node - .content_id() - .map_err(|e| GraphError::NodeOperation(e.to_string()))?; - node_map.insert(node_cid, node.parents().to_vec()); - } - Err(e) => { - println!("Error deserializing node: {e}"); - } - } + let node = Node::::from_bytes(&value) + .map_err(|e| GraphError::NodeOperation(e.to_string()))?; + let node_cid = node + .content_id() + .map_err(|e| GraphError::NodeOperation(e.to_string()))?; + node_map.insert(node_cid, node.parents().to_vec()); } iter.advance(); } @@ -136,6 +161,7 @@ mod tests { use std::time::{SystemTime, UNIX_EPOCH}; use tempfile::tempdir; + /// Creates a simple test node helper. fn create_test_node(payload: &str) -> Node { let timestamp = SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/src/lib.rs b/src/lib.rs index b1ff0b2..0e98d05 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,3 +4,4 @@ pub mod dasl; pub mod graph; pub mod masl; pub mod repo; +pub mod storage; diff --git a/src/repo.rs b/src/repo.rs index f621d2a..ac06aee 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -3,6 +3,7 @@ use crate::convergence::{ resolver::ConflictResolver, }; use crate::crdt::error::{CrdtError, Result}; +use crate::storage::{BatchError, LeveldbBatchGuard, SharedLeveldb, SharedLeveldbAccess}; use crate::{ crdt::{ crdt_state::CrdtState, @@ -10,17 +11,25 @@ use crate::{ reducer::LwwReducer, storage::OperationStorage, }, + dasl::node::Node, graph::{dag::DagGraph, storage::NodeStorage}, }; use cid::Cid; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::fmt::Debug; +use std::rc::Rc; + +struct PendingNode { + cid: Cid, + parents: Vec, + metadata: ContentMetadata, +} pub struct Repo where - OpStore: OperationStorage, - NodeStore: NodeStorage, + OpStore: OperationStorage + SharedLeveldbAccess, + NodeStore: NodeStorage + SharedLeveldbAccess, Payload: Clone + Serialize + for<'de> Deserialize<'de> + Debug, { pub state: CrdtState, @@ -30,8 +39,8 @@ where impl Repo where - OpStore: OperationStorage, - NodeStore: NodeStorage, + OpStore: OperationStorage + SharedLeveldbAccess, + NodeStore: NodeStorage + SharedLeveldbAccess, Payload: Clone + Serialize + for<'de> Deserialize<'de> + Debug, { pub fn new( @@ -56,82 +65,133 @@ where self.commit_operation_internal(op, false) } + pub fn latest(&self, genesis_id: &Cid) -> Option { + self.dag.calculate_latest(genesis_id).ok().flatten() + } + + /// Convenience wrapper around `DagGraph::get_genesis` + pub fn get_genesis(&self, cid: &Cid) -> Result { + self.dag.get_genesis(cid).map_err(CrdtError::Graph) + } + + pub fn get_operations_with_index( + &self, + genesis: &Cid, + ) -> Result)>> { + let mut ops = self.state.get_operations_by_genesis(genesis)?; + ops.sort_by_key(|op| op.timestamp); + Ok(ops + .into_iter() + .enumerate() + .map(|(idx, op)| (idx + 1, op)) + .collect()) + } + + /// Return parent -> children adjacency for the specified genesis (DAG structure). + pub fn branching_history(&self, genesis: &Cid) -> Result>> { + let nodes = self + .dag + .get_nodes_by_genesis(genesis) + .map_err(CrdtError::Graph)?; + + let mut adjacency: HashMap> = HashMap::new(); + for &cid in &nodes { + if let Some(node) = self.dag.get_node(&cid).map_err(CrdtError::Graph)? { + for parent in node.parents() { + adjacency.entry(*parent).or_default().insert(cid); + } + adjacency.entry(cid).or_default(); + } + } + + Ok(adjacency + .into_iter() + .map(|(cid, set)| { + let mut children: Vec = set.into_iter().collect(); + children.sort(); + (cid, children) + }) + .collect()) + } + + /// Find a linear path from genesis to the latest head. + pub fn linear_history(&self, genesis: &Cid) -> Result> { + let adjacency = self.branching_history(genesis)?; + let mut path = Vec::new(); + let mut current = *genesis; + let mut visited = HashSet::new(); + + while visited.insert(current) { + path.push(current); + let children = adjacency.get(¤t).cloned().unwrap_or_default(); + + if children.is_empty() { + break; + } + + let mut best: Option<(Cid, (bool, u64))> = None; + for child in children { + let info = self.node_characteristics(&child)?; + if let Some((_, best_info)) = &best { + if info > *best_info { + best = Some((child, info)); + } + } else { + best = Some((child, info)); + } + } + + let Some((next, _)) = best else { + break; + }; + + current = next; + } + + Ok(path) + } + + fn shared_leveldb(&self) -> Result> { + let op_db = self.state.storage().shared_leveldb().ok_or_else(|| { + CrdtError::Internal("operation storage does not support batching".into()) + })?; + let node_db = + self.dag.storage.shared_leveldb().ok_or_else(|| { + CrdtError::Internal("node storage does not support batching".into()) + })?; + + if !Rc::ptr_eq(&op_db, &node_db) { + return Err(CrdtError::Internal( + "operation and node storage must share the same LevelDB instance for transactions" + .into(), + )); + } + + Ok(op_db) + } + fn commit_operation_internal( &mut self, op: Operation, skip_auto_merge: bool, ) -> Result { let mut op = op; + let shared = self.shared_leveldb()?; + let batch_guard = Self::begin_shared_batch(&shared)?; + let mut pending_nodes: Vec = Vec::new(); if !skip_auto_merge { - match &op.kind { - OperationType::Update(_) | OperationType::Delete => { - if op.parents.is_empty() { - let merged_head = self - .check_and_merge(&op.genesis)? - .or_else(|| self.dag.calculate_latest(&op.genesis).ok().flatten()) - .ok_or_else(|| { - CrdtError::Internal(format!( - "No head available for genesis {} to attach operation", - op.genesis - )) - })?; - - op.parents = vec![merged_head]; - } else { - self.validate_parent_genesis(&op.genesis, &op.parents)?; - } - } - OperationType::Merge(_) => { - if op.parents.is_empty() { - op.parents = self.find_heads(&op.genesis)?; - } - self.validate_parent_genesis(&op.genesis, &op.parents)?; - } - OperationType::Create(_) => {} - } + self.ensure_parent_context(&mut op, &mut pending_nodes)?; } - let cid = match &op.kind { + let cid = match op.kind.clone() { OperationType::Create(payload) => { - let genesis_cid = self - .dag - .add_genesis_node(payload.clone(), ContentMetadata::default())?; - op.genesis = genesis_cid; - genesis_cid + self.stage_create(payload, &mut op, &mut pending_nodes)? } - OperationType::Update(payload) => self.dag.add_child_node( - payload.clone(), - op.parents.clone(), - op.genesis, - self.resolve_metadata_for_commit(&op.genesis, &op.parents)?, - )?, - OperationType::Delete => { - let ops = self.state.get_operations_by_genesis(&op.genesis)?; - let last_payload = ops - .iter() - .filter_map(|operation| { - operation - .payload() - .cloned() - .map(|payload| (operation.timestamp, payload)) - }) - .max_by_key(|(timestamp, _)| *timestamp) - .map(|(_, payload)| payload) - .ok_or_else(|| { - CrdtError::Internal(format!( - "content must exist for delete operation: {}", - op.genesis - )) - })?; - - self.dag.add_child_node( - last_payload, - op.parents.clone(), - op.genesis, - self.resolve_metadata_for_commit(&op.genesis, &op.parents)?, - )? + OperationType::Update(payload) => { + self.stage_update(payload, &op, &mut pending_nodes)? } + OperationType::Delete => self.stage_delete(&op, &mut pending_nodes)?, OperationType::Merge(_) => { return Err(CrdtError::Internal( "Merge operations must be committed via auto-merge".to_string(), @@ -139,30 +199,155 @@ where } }; - self.state.apply(op)?; + if let Err(err) = self.state.apply(op) { + self.rollback_pending_nodes(&pending_nodes); + return Err(err); + } + + if let Err(status) = batch_guard.commit() { + self.rollback_pending_nodes(&pending_nodes); + return Err(CrdtError::Storage(status)); + } Ok(cid) } - pub fn latest(&self, genesis_id: &Cid) -> Option { - self.dag.calculate_latest(genesis_id).ok().flatten() + fn begin_shared_batch(shared: &SharedLeveldb) -> Result> { + shared.begin_batch().map_err(|err| match err { + BatchError::Unsupported => CrdtError::Internal( + "current storage backend does not support transactions".to_string(), + ), + BatchError::AlreadyActive => CrdtError::Internal( + "a transaction is already active on the shared LevelDB".to_string(), + ), + BatchError::Commit(status) => CrdtError::Storage(status), + }) } - /// Convenience wrapper around `DagGraph::get_genesis` - pub fn get_genesis(&self, cid: &Cid) -> Result { - self.dag.get_genesis(cid).map_err(CrdtError::Graph) + + fn rollback_pending_nodes(&mut self, pending: &[PendingNode]) { + for node in pending.iter().rev() { + self.dag.rollback_pending_node(&node.cid, &node.parents); + } } - pub fn get_operations_with_index( - &self, - genesis: &Cid, - ) -> Result)>> { - let mut ops = self.state.get_operations_by_genesis(genesis)?; - ops.sort_by_key(|op| op.timestamp); - Ok(ops - .into_iter() - .enumerate() - .map(|(idx, op)| (idx + 1, op)) - .collect()) + fn ensure_parent_context( + &mut self, + op: &mut Operation, + pending_nodes: &mut Vec, + ) -> Result<()> { + match &op.kind { + OperationType::Update(_) | OperationType::Delete => { + if op.parents.is_empty() { + let merged_head = self + .check_and_merge(&op.genesis, pending_nodes)? + .or_else(|| self.dag.calculate_latest(&op.genesis).ok().flatten()) + .ok_or_else(|| { + CrdtError::Internal(format!( + "No head available for genesis {} to attach operation", + op.genesis + )) + })?; + + op.parents = vec![merged_head]; + } else { + self.validate_parent_genesis(&op.genesis, &op.parents)?; + } + } + OperationType::Merge(_) => { + if op.parents.is_empty() { + op.parents = self.find_heads(&op.genesis)?; + } + self.validate_parent_genesis(&op.genesis, &op.parents)?; + } + OperationType::Create(_) => {} + } + Ok(()) + } + + fn stage_create( + &mut self, + payload: Payload, + op: &mut Operation, + pending_nodes: &mut Vec, + ) -> Result { + let (genesis_cid, node) = self + .dag + .prepare_genesis_node(payload, ContentMetadata::default())?; + let cid = self.stage_prepared_node(genesis_cid, node, pending_nodes)?; + op.genesis = cid; + Ok(cid) + } + + fn stage_update( + &mut self, + payload: Payload, + op: &Operation, + pending_nodes: &mut Vec, + ) -> Result { + let metadata = + self.resolve_metadata_for_commit(&op.genesis, &op.parents, pending_nodes.as_slice())?; + let (cid, node) = + self.dag + .prepare_child_node(payload, op.parents.clone(), op.genesis, metadata)?; + self.stage_prepared_node(cid, node, pending_nodes) + } + + fn stage_delete( + &mut self, + op: &Operation, + pending_nodes: &mut Vec, + ) -> Result { + let ops = self.state.get_operations_by_genesis(&op.genesis)?; + let last_payload = ops + .iter() + .filter_map(|operation| { + operation + .payload() + .cloned() + .map(|payload| (operation.timestamp, payload)) + }) + .max_by_key(|(timestamp, _)| *timestamp) + .map(|(_, payload)| payload) + .ok_or_else(|| { + CrdtError::Internal(format!( + "content must exist for delete operation: {}", + op.genesis + )) + })?; + + let metadata = + self.resolve_metadata_for_commit(&op.genesis, &op.parents, pending_nodes.as_slice())?; + let (cid, node) = + self.dag + .prepare_child_node(last_payload, op.parents.clone(), op.genesis, metadata)?; + self.stage_prepared_node(cid, node, pending_nodes) + } + + fn stage_prepared_node( + &mut self, + cid: Cid, + node: Node, + pending_nodes: &mut Vec, + ) -> Result { + let pending = self.persist_prepared_node(cid, &node)?; + pending_nodes.push(pending); + Ok(cid) + } + + fn persist_prepared_node( + &mut self, + cid: Cid, + node: &Node, + ) -> Result { + self.dag.storage.put(node).map_err(CrdtError::Graph)?; + self.dag + .register_prepared_node(cid, node) + .map_err(CrdtError::Graph)?; + Ok(PendingNode { + cid, + parents: node.parents().to_vec(), + metadata: node.metadata().clone(), + }) } /// Get the latest parent nodes for the given genesis @@ -178,7 +363,11 @@ where Ok(()) } - fn check_and_merge(&mut self, genesis: &Cid) -> Result> { + fn check_and_merge( + &mut self, + genesis: &Cid, + pending_nodes: &mut Vec, + ) -> Result> { let heads = self.find_heads(genesis)?; if heads.len() <= 1 { @@ -199,12 +388,16 @@ where self.validate_parent_genesis(genesis, &heads)?; - let merge_cid = self.dag.add_child_node( - merge_node.payload().clone(), - heads.clone(), - *genesis, - merge_node.metadata().clone(), - )?; + let (merge_cid, node) = self + .dag + .prepare_child_node( + merge_node.payload().clone(), + heads.clone(), + *genesis, + merge_node.metadata().clone(), + ) + .map_err(CrdtError::Graph)?; + let pending = self.persist_prepared_node(merge_cid, &node)?; let mut merge_op = Operation::new( *genesis, @@ -212,7 +405,13 @@ where "auto-merge".to_string(), ); merge_op.parents = heads; - self.state.apply(merge_op)?; + if let Err(err) = self.state.apply(merge_op) { + self.dag + .rollback_pending_node(&pending.cid, &pending.parents); + return Err(err); + } + + pending_nodes.push(pending); Ok(Some(merge_cid)) } @@ -256,8 +455,12 @@ where &self, genesis: &Cid, parents: &[Cid], + pending_nodes: &[PendingNode], ) -> Result { if let Some(parent) = parents.first() { + if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == parent) { + return Ok(pending.metadata.clone()); + } let node = self .dag .get_node(parent) @@ -265,6 +468,9 @@ where .ok_or_else(|| CrdtError::Internal(format!("Parent node not found: {parent}")))?; Ok(node.metadata().clone()) } else { + if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == genesis) { + return Ok(pending.metadata.clone()); + } let genesis_node = self .dag .get_node(genesis) @@ -273,70 +479,6 @@ where Ok(genesis_node.metadata().clone()) } } - /// Return parent -> children adjacency for the specified genesis (DAG structure). - pub fn branching_history(&self, genesis: &Cid) -> Result>> { - let nodes = self - .dag - .get_nodes_by_genesis(genesis) - .map_err(CrdtError::Graph)?; - - let mut adjacency: HashMap> = HashMap::new(); - for &cid in &nodes { - if let Some(node) = self.dag.get_node(&cid).map_err(CrdtError::Graph)? { - for parent in node.parents() { - adjacency.entry(*parent).or_default().insert(cid); - } - adjacency.entry(cid).or_default(); - } - } - - Ok(adjacency - .into_iter() - .map(|(cid, set)| { - let mut children: Vec = set.into_iter().collect(); - children.sort(); - (cid, children) - }) - .collect()) - } - - /// Find a linear path from genesis to the latest head. - pub fn linear_history(&self, genesis: &Cid) -> Result> { - let adjacency = self.branching_history(genesis)?; - let mut path = Vec::new(); - let mut current = *genesis; - let mut visited = HashSet::new(); - - while visited.insert(current) { - path.push(current); - let children = adjacency.get(¤t).cloned().unwrap_or_default(); - - if children.is_empty() { - break; - } - - let mut best: Option<(Cid, (bool, u64))> = None; - for child in children { - let info = self.node_characteristics(&child)?; - if let Some((_, best_info)) = &best { - if info > *best_info { - best = Some((child, info)); - } - } else { - best = Some((child, info)); - } - } - - let Some((next, _)) = best else { - break; - }; - - current = next; - } - - Ok(path) - } - fn node_characteristics(&self, cid: &Cid) -> Result<(bool, u64)> { let node = self .dag @@ -354,7 +496,9 @@ mod tests { use crate::crdt::operation::{Operation, OperationType}; use crate::crdt::storage::LeveldbStorage; use crate::graph::storage::LeveldbNodeStorage; + use std::cell::Cell; use tempfile::tempdir; + use ulid::Ulid; #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(transparent)] @@ -368,8 +512,9 @@ mod tests { fn setup_test_repo() -> (TestRepo, tempfile::TempDir) { let dir = tempdir().unwrap(); - let op_storage = LeveldbStorage::open(dir.path().join("ops")).unwrap(); - let node_storage = LeveldbNodeStorage::open(dir.path().join("nodes")); + let shared = SharedLeveldb::open(dir.path().join("store")).unwrap(); + let op_storage = LeveldbStorage::new(shared.clone()); + let node_storage = LeveldbNodeStorage::new(shared); let state = CrdtState::new(op_storage); let dag = DagGraph::new(node_storage); let repo = Repo::new(state, dag); @@ -387,6 +532,62 @@ mod tests { std::thread::sleep(std::time::Duration::from_millis(1)); } + struct FailingOperationStorage { + inner: S, + fail_next: Cell, + } + + impl FailingOperationStorage { + fn fail_on_first(inner: S) -> Self { + Self { + inner, + fail_next: Cell::new(true), + } + } + } + + impl OperationStorage for FailingOperationStorage + where + S: OperationStorage, + { + fn save_operation(&self, op: &Operation) -> crate::crdt::error::Result<()> { + if self.fail_next.replace(false) { + Err(CrdtError::Internal( + "forced failure for testing".to_string(), + )) + } else { + self.inner.save_operation(op) + } + } + + fn load_operations( + &self, + genesis: &ContentId, + ) -> crate::crdt::error::Result>> { + self.inner.load_operations(genesis) + } + + fn get_operation( + &self, + op_id: &Ulid, + ) -> crate::crdt::error::Result>> { + self.inner.get_operation(op_id) + } + + fn delete_operation(&self, op_id: &Ulid) -> crate::crdt::error::Result<()> { + self.inner.delete_operation(op_id) + } + } + + impl SharedLeveldbAccess for FailingOperationStorage + where + S: SharedLeveldbAccess, + { + fn shared_leveldb(&self) -> Option> { + self.inner.shared_leveldb() + } + } + #[test] fn test_create_operation() { let (mut repo, _) = setup_test_repo(); @@ -428,6 +629,40 @@ mod tests { assert_ne!(create_cid, update_cid); } + #[test] + fn test_create_operation_rolls_back_on_state_failure() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path().join("store")).unwrap(); + let op_storage = + FailingOperationStorage::fail_on_first(LeveldbStorage::new(shared.clone())); + let node_storage = LeveldbNodeStorage::new(shared); + let state = CrdtState::new(op_storage); + let dag = DagGraph::new(node_storage); + let mut repo = Repo::new(state, dag); + + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"rollback-test").unwrap(), + ); + let op = make_test_operation( + initial_genesis, + OperationType::Create(TestPayload("should not persist".to_string())), + ); + let op_id = op.id; + + let result = repo.commit_operation(op); + assert!(result.is_err()); + + let node_map = repo.dag.storage.get_node_map().unwrap(); + assert!( + node_map.is_empty(), + "expected DAG to be empty after rollback, found {node_map:?}" + ); + assert!( + repo.state.get_operation(&op_id).unwrap().is_none(), + "operation was persisted despite failure" + ); + } #[test] fn test_update_with_explicit_parent_is_respected() { let (mut repo, _) = setup_test_repo(); diff --git a/src/storage/mod.rs b/src/storage/mod.rs new file mode 100644 index 0000000..b36909c --- /dev/null +++ b/src/storage/mod.rs @@ -0,0 +1,3 @@ +mod shared_leveldb; + +pub use shared_leveldb::{BatchError, LeveldbBatchGuard, SharedLeveldb, SharedLeveldbAccess}; diff --git a/src/storage/shared_leveldb.rs b/src/storage/shared_leveldb.rs new file mode 100644 index 0000000..0a9cc46 --- /dev/null +++ b/src/storage/shared_leveldb.rs @@ -0,0 +1,160 @@ +use rusty_leveldb::{Options, Status, WriteBatch, DB as Database}; +use std::cell::RefCell; +use std::path::Path; +use std::rc::Rc; + +#[derive(Debug)] +pub enum BatchError { + Unsupported, + AlreadyActive, + Commit(Status), +} + +pub struct SharedLeveldb { + db: RefCell, + active_batch: RefCell>, +} + +impl SharedLeveldb { + pub fn open>(path: P) -> Result, Status> { + let opts = Options { + create_if_missing: true, + ..Default::default() + }; + let db = Database::open(path, opts)?; + Ok(Rc::new(Self { + db: RefCell::new(db), + active_batch: RefCell::new(None), + })) + } + + pub fn begin_batch(&self) -> Result, BatchError> { + let mut slot = self.active_batch.borrow_mut(); + if slot.is_some() { + return Err(BatchError::AlreadyActive); + } + *slot = Some(WriteBatch::default()); + Ok(LeveldbBatchGuard { + shared: self, + committed: false, + }) + } + + fn commit_batch(&self) -> Result<(), Status> { + let mut slot = self.active_batch.borrow_mut(); + let Some(batch) = slot.take() else { + return Ok(()); + }; + self.db.borrow_mut().write(batch, true) + } + + fn abort_batch(&self) { + self.active_batch.borrow_mut().take(); + } + + pub fn with_active_batch(&self, f: F) -> Option + where + F: FnOnce(&mut WriteBatch) -> R, + { + let mut slot = self.active_batch.borrow_mut(); + slot.as_mut().map(f) + } + + pub fn db(&self) -> &RefCell { + &self.db + } +} + +pub struct LeveldbBatchGuard<'a> { + shared: &'a SharedLeveldb, + committed: bool, +} + +impl<'a> LeveldbBatchGuard<'a> { + pub fn commit(mut self) -> Result<(), Status> { + self.shared.commit_batch()?; + self.committed = true; + Ok(()) + } +} + +impl Drop for LeveldbBatchGuard<'_> { + fn drop(&mut self) { + if !self.committed { + self.shared.abort_batch(); + } + } +} + +pub trait SharedLeveldbAccess { + fn shared_leveldb(&self) -> Option>; +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn begin_batch_prevents_nested_batches() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path()).expect("open shared db"); + + let guard = shared.begin_batch().expect("begin first batch"); + match shared.begin_batch() { + Err(BatchError::AlreadyActive) => {} + Ok(_) => panic!("expected AlreadyActive error, got Ok"), + Err(err) => panic!("unexpected batch error: {err:?}"), + } + drop(guard); + + shared + .begin_batch() + .expect("batch should be available after guard drop") + .commit() + .expect("commit empty batch"); + } + + #[test] + fn commit_batch_persists_operations() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path()).expect("open shared db"); + + let guard = shared.begin_batch().expect("begin batch"); + let key = b"test-key"; + let value = b"test-value"; + + let inserted = shared.with_active_batch(|batch| batch.put(key, value)); + assert!(inserted.is_some(), "expected active batch to exist"); + + guard.commit().expect("commit batch"); + + let stored = shared + .db() + .borrow_mut() + .get(key) + .expect("value should exist after commit"); + assert_eq!(stored.as_slice(), value); + } + + #[test] + fn dropping_guard_discards_pending_operations() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path()).expect("open shared db"); + let key = b"discard-key"; + let value = b"discard-value"; + + { + let _guard = shared.begin_batch().expect("begin batch"); + let inserted = shared.with_active_batch(|batch| batch.put(key, value)); + assert!(inserted.is_some(), "expected active batch to exist"); + // guard dropped here without commit + } + + let result = shared.db().borrow_mut().get(key); + assert!( + result.is_none(), + "value should not be persisted when batch guard is dropped without commit" + ); + } +} From f1a5e70bb4dfe02d33674276cfbed1279f7e8c7b Mon Sep 17 00:00:00 2001 From: Yu-da-1 Date: Sat, 15 Nov 2025 15:17:34 +0900 Subject: [PATCH 2/2] feat: add test case --- src/repo.rs | 288 ++++++++++++++++++++++++++++++++++ src/storage/shared_leveldb.rs | 15 ++ 2 files changed, 303 insertions(+) diff --git a/src/repo.rs b/src/repo.rs index ac06aee..807dc0b 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -495,7 +495,9 @@ mod tests { use super::*; use crate::crdt::operation::{Operation, OperationType}; use crate::crdt::storage::LeveldbStorage; + use crate::graph::error::GraphError; use crate::graph::storage::LeveldbNodeStorage; + use rusty_leveldb::{Status, StatusCode}; use std::cell::Cell; use tempfile::tempdir; use ulid::Ulid; @@ -538,12 +540,23 @@ mod tests { } impl FailingOperationStorage { + fn new(inner: S) -> Self { + Self { + inner, + fail_next: Cell::new(false), + } + } + fn fail_on_first(inner: S) -> Self { Self { inner, fail_next: Cell::new(true), } } + + fn fail_on_next(&self) { + self.fail_next.set(true); + } } impl OperationStorage for FailingOperationStorage @@ -588,6 +601,56 @@ mod tests { } } + struct FailingNodeStorage { + inner: S, + fail_next_put: Cell, + } + + impl FailingNodeStorage { + fn fail_on_first_put(inner: S) -> Self { + Self { + inner, + fail_next_put: Cell::new(true), + } + } + } + + impl NodeStorage for FailingNodeStorage + where + S: NodeStorage, + { + fn get(&self, content_id: &Cid) -> crate::graph::error::Result>> { + self.inner.get(content_id) + } + + fn put(&self, node: &Node) -> crate::graph::error::Result<()> { + if self.fail_next_put.replace(false) { + Err(GraphError::Internal( + "injected node storage failure".to_string(), + )) + } else { + self.inner.put(node) + } + } + + fn delete(&self, content_id: &Cid) -> crate::graph::error::Result<()> { + self.inner.delete(content_id) + } + + fn get_node_map(&self) -> crate::graph::error::Result>> { + self.inner.get_node_map() + } + } + + impl SharedLeveldbAccess for FailingNodeStorage + where + S: SharedLeveldbAccess, + { + fn shared_leveldb(&self) -> Option> { + self.inner.shared_leveldb() + } + } + #[test] fn test_create_operation() { let (mut repo, _) = setup_test_repo(); @@ -604,6 +667,45 @@ mod tests { assert_eq!(repo.latest(&cid).unwrap(), cid); } + #[test] + fn test_create_operation_fails_when_node_storage_errors() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path().join("store")).unwrap(); + let op_storage = LeveldbStorage::new(shared.clone()); + let node_storage = + FailingNodeStorage::fail_on_first_put(LeveldbNodeStorage::new(shared.clone())); + let state = CrdtState::new(op_storage); + let dag = DagGraph::new(node_storage); + let mut repo = Repo::new(state, dag); + + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"create-fail").unwrap(), + ); + let op = make_test_operation( + initial_genesis, + OperationType::Create(TestPayload("should fail".to_string())), + ); + let op_id = op.id; + + let err = repo.commit_operation(op).unwrap_err(); + match err { + CrdtError::Graph(GraphError::Internal(message)) => { + assert!(message.contains("injected node storage failure")); + } + other => panic!("unexpected error: {other:?}"), + } + + assert!( + repo.state.get_operation(&op_id).unwrap().is_none(), + "operation should not be persisted on failure" + ); + assert!( + repo.dag.storage.get_node_map().unwrap().is_empty(), + "dag should remain empty when node storage fails" + ); + } + #[test] fn test_update_operation() { let (mut repo, _) = setup_test_repo(); @@ -629,6 +731,36 @@ mod tests { assert_ne!(create_cid, update_cid); } + #[test] + fn test_update_operation_without_existing_head_fails() { + let (mut repo, _) = setup_test_repo(); + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"update-no-head").unwrap(), + ); + let op = make_test_operation( + initial_genesis, + OperationType::Update(TestPayload("orphaned".to_string())), + ); + + let err = repo.commit_operation(op).unwrap_err(); + match err { + CrdtError::Internal(message) => { + assert!(message.contains("No head available")); + } + other => panic!("unexpected error: {other:?}"), + } + + let stored_ops = repo + .state + .get_operations_by_genesis(&initial_genesis) + .unwrap(); + assert!( + stored_ops.is_empty(), + "update should not persist when no head exists" + ); + } + #[test] fn test_create_operation_rolls_back_on_state_failure() { let dir = tempdir().unwrap(); @@ -663,6 +795,119 @@ mod tests { "operation was persisted despite failure" ); } + + #[test] + fn test_create_operation_rolls_back_when_batch_commit_fails() { + let (mut repo, _) = setup_test_repo(); + let shared = repo + .state + .storage() + .shared_leveldb() + .expect("shared leveldb instance"); + shared.inject_commit_failure(Status::new(StatusCode::IOError, "forced commit failure")); + + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"batch-failure").unwrap(), + ); + let op = make_test_operation( + initial_genesis, + OperationType::Create(TestPayload("batch-fail".to_string())), + ); + let op_id = op.id; + + let err = repo.commit_operation(op).unwrap_err(); + match err { + CrdtError::Storage(status) => { + assert_eq!(status.code, StatusCode::IOError); + assert!(status.err.contains("forced commit failure")); + } + other => panic!("unexpected error: {other:?}"), + } + + assert!( + repo.state.get_operation(&op_id).unwrap().is_none(), + "operation should not persist when batch commit fails" + ); + assert!( + repo.dag.storage.get_node_map().unwrap().is_empty(), + "dag should be rolled back when batch commit fails" + ); + } + + #[test] + fn test_rollback_pending_nodes_restores_heads_after_failure() { + let dir = tempdir().unwrap(); + let shared = SharedLeveldb::open(dir.path().join("store")).unwrap(); + let op_storage = FailingOperationStorage::new(LeveldbStorage::new(shared.clone())); + let node_storage = LeveldbNodeStorage::new(shared); + let state = CrdtState::new(op_storage); + let dag = DagGraph::new(node_storage); + let mut repo = Repo::new(state, dag); + + let seed = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"rollback-pending").unwrap(), + ); + let create = make_test_operation(seed, OperationType::Create(TestPayload("root".into()))); + let genesis = repo.commit_operation(create).unwrap(); + + let mut branch1 = make_test_operation( + genesis, + OperationType::Update(TestPayload("branch-1".into())), + ); + branch1.parents.push(genesis); + let branch1_cid = repo.commit_operation(branch1).unwrap(); + sleep_for_ordering(); + + let mut branch2 = make_test_operation( + genesis, + OperationType::Update(TestPayload("branch-2".into())), + ); + branch2.parents.push(genesis); + let branch2_cid = repo.commit_operation(branch2).unwrap(); + + let original_heads = repo.find_heads(&genesis).unwrap(); + assert_eq!(original_heads.len(), 2); + assert!(original_heads.contains(&branch1_cid)); + assert!(original_heads.contains(&branch2_cid)); + + repo.state.storage().fail_on_next(); + + let update = make_test_operation( + genesis, + OperationType::Update(TestPayload("should-rollback".into())), + ); + let err = repo.commit_operation(update).unwrap_err(); + match err { + CrdtError::Internal(message) => { + assert!(message.contains("forced failure for testing")); + } + other => panic!("unexpected error: {other:?}"), + } + + let heads_after = repo.find_heads(&genesis).unwrap(); + assert_eq!(heads_after.len(), 2); + assert!(heads_after.contains(&branch1_cid)); + assert!(heads_after.contains(&branch2_cid)); + + let ops = repo.state.get_operations_by_genesis(&genesis).unwrap(); + assert_eq!( + ops.len(), + 3, + "rollback should leave only the original create and two branch updates" + ); + + let node_map = repo.dag.storage.get_node_map().unwrap(); + assert!(node_map.contains_key(&genesis)); + assert!(node_map.contains_key(&branch1_cid)); + assert!(node_map.contains_key(&branch2_cid)); + assert_eq!( + node_map.len(), + 3, + "no additional DAG nodes should remain after rollback" + ); + } #[test] fn test_update_with_explicit_parent_is_respected() { let (mut repo, _) = setup_test_repo(); @@ -804,6 +1049,49 @@ mod tests { assert_ne!(create_cid, delete_cid); } + #[test] + fn test_delete_operation_without_existing_payload_fails() { + let (mut repo, _) = setup_test_repo(); + let (genesis_cid, genesis_node) = repo + .dag + .prepare_genesis_node( + TestPayload("dangling".to_string()), + ContentMetadata::default(), + ) + .unwrap(); + repo.dag.storage.put(&genesis_node).unwrap(); + repo.dag + .register_prepared_node(genesis_cid, &genesis_node) + .unwrap(); + + let op = make_test_operation(genesis_cid, OperationType::Delete); + let op_id = op.id; + + let err = repo.commit_operation(op).unwrap_err(); + match err { + CrdtError::Internal(message) => { + assert!(message.contains("content must exist")); + } + other => panic!("unexpected error: {other:?}"), + } + + assert!( + repo.state.get_operation(&op_id).unwrap().is_none(), + "delete operation should not be stored when payload is missing" + ); + assert!( + repo.state + .get_operations_by_genesis(&genesis_cid) + .unwrap() + .is_empty(), + "operation history should remain empty on failure" + ); + assert!( + repo.dag.get_node(&genesis_cid).unwrap().is_some(), + "existing genesis node should remain after failed delete" + ); + } + #[test] fn test_multiple_genesis_entries() { let (mut repo, _) = setup_test_repo(); diff --git a/src/storage/shared_leveldb.rs b/src/storage/shared_leveldb.rs index 0a9cc46..12b112f 100644 --- a/src/storage/shared_leveldb.rs +++ b/src/storage/shared_leveldb.rs @@ -13,6 +13,8 @@ pub enum BatchError { pub struct SharedLeveldb { db: RefCell, active_batch: RefCell>, + #[cfg(test)] + commit_fail_status: RefCell>, } impl SharedLeveldb { @@ -25,6 +27,8 @@ impl SharedLeveldb { Ok(Rc::new(Self { db: RefCell::new(db), active_batch: RefCell::new(None), + #[cfg(test)] + commit_fail_status: RefCell::new(None), })) } @@ -45,6 +49,10 @@ impl SharedLeveldb { let Some(batch) = slot.take() else { return Ok(()); }; + #[cfg(test)] + if let Some(status) = self.commit_fail_status.borrow_mut().take() { + return Err(status); + } self.db.borrow_mut().write(batch, true) } @@ -90,6 +98,13 @@ pub trait SharedLeveldbAccess { fn shared_leveldb(&self) -> Option>; } +#[cfg(test)] +impl SharedLeveldb { + pub fn inject_commit_failure(&self, status: Status) { + self.commit_fail_status.borrow_mut().replace(status); + } +} + #[cfg(test)] mod tests { use super::*;