From aff4b8e5a51adca34acc9848f61297aed2f53380 Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 30 Nov 2025 04:25:13 +0900 Subject: [PATCH 1/4] Add auto-merge and thread safety features to storage modules --- readme.md | 24 +++++++++++++++++++++-- src/convergence/resolver.rs | 17 ++++++++--------- src/crdt/storage.rs | 33 ++++++++++++++++++++------------ src/graph/dag.rs | 38 ++++++++++++++++++------------------- src/graph/storage.rs | 35 ++++++++++++++++++++-------------- 5 files changed, 91 insertions(+), 56 deletions(-) diff --git a/readme.md b/readme.md index 9a4b45b..47357c9 100644 --- a/readme.md +++ b/readme.md @@ -6,8 +6,10 @@ CRSL is a Rust library for content versioning and CRDT (Conflict-free Replicated - **Content Versioning**: Content creation, update, deletion, and history management - **CRDT Support**: Conflict resolution through Last-Write-Wins (LWW) reducer +- **Auto-Merge**: Automatic conflict resolution when multiple heads exist - **DAG (Directed Acyclic Graph)**: Efficient version history management - **LevelDB Storage**: High-performance persistent storage +- **Thread-Safe**: Safe to use in async/await environments with `Mutex`-based storage - **CID (Content Identifier)**: IPFS-compatible content identifiers ## 🛠️ Usage @@ -105,11 +107,17 @@ crsl-lib/ │ │ ├── crdt_state.rs # CRDT state management │ │ ├── operation.rs # Operation definitions │ │ ├── reducer.rs # LWW reducer -│ │ ├── storage.rs # Operation storage +│ │ ├── storage.rs # Operation storage (thread-safe) │ │ └── error.rs # Error definitions +│ ├── convergence/ # Conflict resolution +│ │ ├── resolver.rs # Merge orchestration +│ │ ├── policy.rs # MergePolicy trait +│ │ ├── policies/ # Policy implementations +│ │ │ └── lww.rs # Last-Write-Wins policy +│ │ └── metadata.rs # Content metadata │ ├── graph/ # DAG graph implementation │ │ ├── dag.rs # DAG graph management -│ │ ├── storage.rs # Node storage +│ │ ├── storage.rs # Node storage (thread-safe) │ │ └── error.rs # Graph errors │ ├── dasl/ # DASL (Distributed Application Storage Layer) │ ├── masl/ # MASL (Multi-Agent Storage Layer) @@ -183,6 +191,11 @@ cargo doc --open - Conflict resolution through LWW reducer - Integration with operation storage +### Convergence (`src/convergence/`) +- **MergePolicy trait**: Customizable merge strategies +- **LwwMergePolicy**: Last-Write-Wins merge implementation +- **ConflictResolver**: Automatic merge node creation + ### DAG Graph (`src/graph/dag.rs`) - DAG management for version history - Node addition, retrieval, and history tracking @@ -191,12 +204,19 @@ cargo doc --open ### Repository (`src/repo.rs`) - Integration of CRDT State and DAG Graph - Operation commit and history management +- Auto-merge when multiple heads exist - High-level API provision ### Operations (`src/crdt/operation.rs`) - Create: New content creation - Update: Content updates - Delete: Content deletion +- Merge: Automatic merge operations + +### Thread Safety +- `LeveldbStorage` and `LeveldbNodeStorage` use `Mutex` internally +- `OperationStorage` and `NodeStorage` traits require `Send + Sync` +- Safe to use with `Arc>` in async/await environments ## 📄 License diff --git a/src/convergence/resolver.rs b/src/convergence/resolver.rs index d875325..fd58424 100644 --- a/src/convergence/resolver.rs +++ b/src/convergence/resolver.rs @@ -117,13 +117,12 @@ mod tests { use crate::graph::storage::NodeStorage; use multihash::Multihash; use serde::{Deserialize, Serialize}; - use std::cell::RefCell; use std::collections::HashMap; - use std::rc::Rc; + use std::sync::{Arc, Mutex}; #[derive(Clone, Default)] struct MemoryNodeStorage { - nodes: Rc>>>, + nodes: Arc>>>, } impl MemoryNodeStorage @@ -135,18 +134,18 @@ mod tests { let cid = node .content_id() .map_err(|e| GraphError::NodeOperation(e.to_string()))?; - self.nodes.borrow_mut().insert(cid, node.clone()); + self.nodes.lock().unwrap().insert(cid, node.clone()); Ok(()) } } impl NodeStorage for MemoryNodeStorage where - P: Clone + Serialize + for<'de> Deserialize<'de>, - M: Clone + Serialize + for<'de> Deserialize<'de>, + P: Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync, + M: Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync, { fn get(&self, content_id: &Cid) -> GraphResult>> { - Ok(self.nodes.borrow().get(content_id).cloned()) + Ok(self.nodes.lock().unwrap().get(content_id).cloned()) } fn put(&self, node: &Node) -> GraphResult<()> { @@ -154,13 +153,13 @@ mod tests { } fn delete(&self, content_id: &Cid) -> GraphResult<()> { - self.nodes.borrow_mut().remove(content_id); + self.nodes.lock().unwrap().remove(content_id); Ok(()) } fn get_node_map(&self) -> GraphResult>> { let mut map = HashMap::new(); - for (cid, node) in self.nodes.borrow().iter() { + for (cid, node) in self.nodes.lock().unwrap().iter() { map.insert(*cid, node.parents().to_vec()); } Ok(map) diff --git a/src/crdt/storage.rs b/src/crdt/storage.rs index 2d2e45c..365deaf 100644 --- a/src/crdt/storage.rs +++ b/src/crdt/storage.rs @@ -2,19 +2,19 @@ use crate::crdt::error::{CrdtError, Result}; use crate::crdt::operation::Operation; use bincode; use rusty_leveldb::{LdbIterator, Options, DB as Database}; -use std::cell::RefCell; use std::marker::PhantomData; use std::path::Path; +use std::sync::Mutex; use ulid::Ulid; -pub trait OperationStorage { +pub trait OperationStorage: Send + Sync { fn save_operation(&self, op: &Operation) -> Result<()>; fn load_operations(&self, genesis: &ContentId) -> Result>>; fn get_operation(&self, op_id: &Ulid) -> Result>>; } pub struct LeveldbStorage { - db: RefCell, + db: Mutex, _marker: PhantomData<(ContentId, T)>, } @@ -26,7 +26,7 @@ impl LeveldbStorage { }; let db = Database::open(path, opts).map_err(CrdtError::Storage)?; Ok(LeveldbStorage { - db: RefCell::new(db), + db: Mutex::new(db), _marker: PhantomData, }) } @@ -41,23 +41,27 @@ impl LeveldbStorage { impl OperationStorage for LeveldbStorage where - ContentId: serde::Serialize + for<'de> serde::Deserialize<'de> + PartialEq + std::fmt::Debug, - T: serde::Serialize + for<'de> serde::Deserialize<'de> + std::fmt::Debug, + ContentId: + serde::Serialize + for<'de> serde::Deserialize<'de> + PartialEq + std::fmt::Debug + Send + Sync, + T: serde::Serialize + for<'de> serde::Deserialize<'de> + std::fmt::Debug + Send + Sync, { 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)?; + self.db + .lock() + .map_err(|e| CrdtError::Internal(format!("Lock poisoned: {}", e)))? + .put(&key, &value)?; Ok(()) } fn load_operations(&self, genesis: &ContentId) -> Result>> { let mut result = Vec::new(); - let mut iter = self + let mut db = self .db - .borrow_mut() - .new_iter() - .map_err(CrdtError::Storage)?; + .lock() + .map_err(|e| CrdtError::Internal(format!("Lock poisoned: {}", e)))?; + let mut iter = db.new_iter().map_err(CrdtError::Storage)?; // todo: Implement efficient search methods iter.seek_to_first(); let mut key = Vec::new(); @@ -81,7 +85,12 @@ 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 + .db + .lock() + .map_err(|e| CrdtError::Internal(format!("Lock poisoned: {}", e)))? + .get(&key) + { Some(raw) => { let (op, _) = bincode::serde::decode_from_slice::, _>( &raw, diff --git a/src/graph/dag.rs b/src/graph/dag.rs index 1bb2c89..88e1c6b 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -407,27 +407,27 @@ where #[cfg(test)] mod tests { use super::*; - use std::cell::RefCell; use std::collections::BTreeMap; + use std::sync::Mutex; type TestDag = DagGraph>; #[derive(Debug)] struct MockStorage { - edges: std::cell::RefCell>>, - timestamps: std::cell::RefCell>, + edges: Mutex>>, + timestamps: Mutex>, } impl MockStorage { fn new() -> Self { Self { - edges: RefCell::new(HashMap::new()), - timestamps: RefCell::new(HashMap::new()), + edges: Mutex::new(HashMap::new()), + timestamps: Mutex::new(HashMap::new()), } } fn setup_graph(&mut self, structure: &[(Cid, Cid)]) { - let mut edges = self.edges.borrow_mut(); - let mut timestamps = self.timestamps.borrow_mut(); + let mut edges = self.edges.lock().unwrap(); + let mut timestamps = self.timestamps.lock().unwrap(); let mut ts = 1; for (parent, child) in structure { @@ -444,11 +444,11 @@ mod tests { impl NodeStorage for MockStorage where - P: Default + serde::Serialize + serde::de::DeserializeOwned, - M: Default + serde::Serialize + serde::de::DeserializeOwned, + P: Default + serde::Serialize + serde::de::DeserializeOwned + Send + Sync, + M: Default + serde::Serialize + serde::de::DeserializeOwned + Send + Sync, { fn get(&self, content_id: &Cid) -> Result>> { - let edges = self.edges.borrow(); + let edges = self.edges.lock().unwrap(); let parents = edges.get(content_id).cloned(); let parents = match parents { @@ -456,7 +456,7 @@ mod tests { None => return Ok(None), }; - let ts = *self.timestamps.borrow().get(content_id).unwrap_or(&0); + let ts = *self.timestamps.lock().unwrap().get(content_id).unwrap_or(&0); fn find_genesis(edges: &HashMap>, cid: &Cid) -> Cid { let mut current = *cid; @@ -468,7 +468,7 @@ mod tests { } current } - let genesis_cid = find_genesis(&self.edges.borrow(), content_id); + let genesis_cid = find_genesis(&edges, content_id); let node = if parents.is_empty() { Node::new_genesis(P::default(), ts, M::default()) @@ -484,8 +484,8 @@ mod tests { let parents = node.parents().to_vec(); let ts = node.timestamp(); - self.edges.borrow_mut().insert(cid, parents); - self.timestamps.borrow_mut().insert(cid, ts); + self.edges.lock().unwrap().insert(cid, parents); + self.timestamps.lock().unwrap().insert(cid, ts); Ok(()) } @@ -495,7 +495,7 @@ mod tests { } fn get_node_map(&self) -> Result>> { - Ok(self.edges.borrow().clone()) + Ok(self.edges.lock().unwrap().clone()) } } @@ -827,7 +827,7 @@ mod tests { fn test_get_genesis_from_genesis_node() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.borrow_mut().entry(genesis_cid).or_default(); + storage.edges.lock().unwrap().entry(genesis_cid).or_default(); let dag = DagGraph::>::new(storage); let result = dag.get_genesis(&genesis_cid); @@ -862,7 +862,7 @@ mod tests { fn test_calculate_latest_genesis_only() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.borrow_mut().entry(genesis_cid).or_default(); + storage.edges.lock().unwrap().entry(genesis_cid).or_default(); let dag = DagGraph::>::new(storage); let result = dag.calculate_latest(&genesis_cid).unwrap(); assert_eq!(result, Some(genesis_cid)); @@ -907,7 +907,7 @@ mod tests { fn test_get_nodes_by_genesis_genesis_only() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.borrow_mut().entry(genesis_cid).or_default(); + storage.edges.lock().unwrap().entry(genesis_cid).or_default(); let dag = DagGraph::>::new(storage); let result = dag.get_nodes_by_genesis(&genesis_cid).unwrap(); assert_eq!(result, vec![genesis_cid]); @@ -935,7 +935,7 @@ mod tests { let v1_cid = create_test_content_id(b"v1"); let unrelated_cid = create_test_content_id(b"unrelated"); storage.setup_graph(&[(genesis1_cid, v1_cid)]); - storage.edges.borrow_mut().entry(unrelated_cid).or_default(); + storage.edges.lock().unwrap().entry(unrelated_cid).or_default(); let dag = DagGraph::>::new(storage); let mut result = dag.get_nodes_by_genesis(&genesis1_cid).unwrap(); result.sort(); diff --git a/src/graph/storage.rs b/src/graph/storage.rs index cb13af5..0386b0f 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -2,12 +2,12 @@ use crate::dasl::node::Node; use crate::graph::error::{GraphError, Result}; use cid::Cid; use rusty_leveldb::{LdbIterator, Options, DB as Database}; -use std::cell::RefCell; use std::collections::HashMap; use std::path::Path; use std::path::PathBuf; +use std::sync::Mutex; -pub trait NodeStorage { +pub trait NodeStorage: Send + Sync { fn get(&self, content_id: &Cid) -> Result>>; fn put(&self, node: &Node) -> Result<()>; fn delete(&self, content_id: &Cid) -> Result<()>; @@ -15,7 +15,7 @@ pub trait NodeStorage { } pub struct LeveldbNodeStorage { - db: RefCell, + db: Mutex, path: PathBuf, _marker: std::marker::PhantomData<(P, M)>, } @@ -28,7 +28,7 @@ impl Clone for LeveldbNodeStorage { }; let db = Database::open(&self.path, opts).expect("Failed to clone database"); Self { - db: RefCell::new(db), + db: Mutex::new(db), path: self.path.clone(), _marker: std::marker::PhantomData, } @@ -43,7 +43,7 @@ impl LeveldbNodeStorage { }; let db = Database::open(path.as_ref(), opts).unwrap(); Self { - db: RefCell::new(db), + db: Mutex::new(db), path: path.as_ref().to_path_buf(), _marker: std::marker::PhantomData, } @@ -58,12 +58,17 @@ impl LeveldbNodeStorage { impl NodeStorage for LeveldbNodeStorage where - P: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone, - M: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone, + P: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync, + M: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync, { fn get(&self, cid: &Cid) -> Result>> { let key = Self::make_key(cid); - match self.db.borrow_mut().get(&key) { + match self + .db + .lock() + .map_err(|e| GraphError::NodeOperation(format!("Lock poisoned: {}", e)))? + .get(&key) + { Some(raw) => { let node = Node::from_bytes(&raw).map_err(|e| GraphError::NodeOperation(e.to_string()))?; @@ -82,7 +87,8 @@ where .map_err(|e| GraphError::NodeOperation(e.to_string()))?; let key = Self::make_key(&cid); self.db - .borrow_mut() + .lock() + .map_err(|e| GraphError::NodeOperation(format!("Lock poisoned: {}", e)))? .put(&key, &bytes) .map_err(GraphError::Storage)?; Ok(()) @@ -91,7 +97,8 @@ where fn delete(&self, cid: &Cid) -> Result<()> { let key = Self::make_key(cid); self.db - .borrow_mut() + .lock() + .map_err(|e| GraphError::NodeOperation(format!("Lock poisoned: {}", e)))? .delete(&key) .map_err(GraphError::Storage)?; Ok(()) @@ -99,11 +106,11 @@ where fn get_node_map(&self) -> Result>> { let mut node_map = HashMap::new(); - let mut iter = self + let mut db = self .db - .borrow_mut() - .new_iter() - .map_err(GraphError::Storage)?; + .lock() + .map_err(|e| GraphError::NodeOperation(format!("Lock poisoned: {}", e)))?; + let mut iter = db.new_iter().map_err(GraphError::Storage)?; iter.seek_to_first(); let mut key = Vec::new(); let mut value = Vec::new(); From 557f3bc9879da957cb0f5758855f3ade75f9c93e Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 30 Nov 2025 05:01:59 +0900 Subject: [PATCH 2/4] feat: make storage layer thread-safe with Mutex - Replace RefCell with Mutex in SharedLeveldb for thread safety - Replace Rc with Arc for shared ownership across threads - Add Send + Sync bounds to OperationStorage and NodeStorage traits - Update all storage implementations to use Mutex::lock() - Add LockPoisoned variant to BatchError for proper error handling - Update test mocks (MockStorage, MemoryNodeStorage) to use Arc - Update FailingOperationStorage/FailingNodeStorage to use AtomicBool This enables crsl-lib to be safely used in async/await environments and multi-threaded contexts. --- src/crdt/storage.rs | 17 +++++----- src/graph/dag.rs | 2 +- src/graph/storage.rs | 15 ++++---- src/repo.rs | 35 +++++++++++-------- src/storage/shared_leveldb.rs | 64 +++++++++++++++++++++++------------ 5 files changed, 78 insertions(+), 55 deletions(-) diff --git a/src/crdt/storage.rs b/src/crdt/storage.rs index 13fd732..1877ba5 100644 --- a/src/crdt/storage.rs +++ b/src/crdt/storage.rs @@ -5,11 +5,11 @@ use bincode; use rusty_leveldb::LdbIterator; use std::marker::PhantomData; use std::path::Path; -use std::rc::Rc; +use std::sync::Arc; use ulid::Ulid; /// Abstraction over the persistent storage used by `CrdtState`. -pub trait OperationStorage { +pub trait OperationStorage: Send + Sync { fn save_operation(&self, op: &Operation) -> Result<()>; fn load_operations(&self, genesis: &ContentId) -> Result>>; fn get_operation(&self, op_id: &Ulid) -> Result>>; @@ -22,7 +22,7 @@ pub trait OperationStorage { /// LevelDB-backed implementation of [`OperationStorage`]. #[derive(Clone)] pub struct LeveldbStorage { - shared: Rc, + shared: Arc, _marker: PhantomData<(ContentId, T)>, } @@ -32,7 +32,7 @@ impl LeveldbStorage { Ok(Self::new(shared)) } - pub fn new(shared: Rc) -> Self { + pub fn new(shared: Arc) -> Self { Self { shared, _marker: PhantomData, @@ -64,7 +64,7 @@ impl LeveldbStorage { .with_active_batch(|batch| batch.put(key, value)) .is_none() { - self.shared.db().borrow_mut().put(key, value)?; + self.shared.db().put(key, value)?; } Ok(()) } @@ -76,14 +76,14 @@ impl LeveldbStorage { .with_active_batch(|batch| batch.delete(key)) .is_none() { - self.shared.db().borrow_mut().delete(key)?; + self.shared.db().delete(key)?; } Ok(()) } } impl SharedLeveldbAccess for LeveldbStorage { - fn shared_leveldb(&self) -> Option> { + fn shared_leveldb(&self) -> Option> { Some(self.shared.clone()) } } @@ -109,7 +109,6 @@ where let mut iter = self .shared .db() - .borrow_mut() .new_iter() .map_err(CrdtError::Storage)?; iter.seek_to_first(); @@ -134,7 +133,7 @@ where fn get_operation(&self, op_id: &Ulid) -> Result>> { let key = Self::make_key(op_id); - match self.shared.db().borrow_mut().get(&key) { + match self.shared.db().get(&key) { Some(raw) => { let (op, _) = bincode::serde::decode_from_slice::, _>( &raw, diff --git a/src/graph/dag.rs b/src/graph/dag.rs index 3140e1f..7fdbe00 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -441,8 +441,8 @@ where mod tests { use super::*; use crate::graph::storage::LeveldbNodeStorage; - use std::cell::RefCell; use std::collections::BTreeMap; + use std::sync::Mutex; use tempfile::tempdir; type TestDag = DagGraph>; diff --git a/src/graph/storage.rs b/src/graph/storage.rs index d1f2ba3..6f511aa 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -5,10 +5,10 @@ use cid::Cid; use rusty_leveldb::LdbIterator; use std::collections::HashMap; use std::path::Path; -use std::rc::Rc; +use std::sync::Arc; /// Minimal interface required for persisting DAG nodes. -pub trait NodeStorage { +pub trait NodeStorage: Send + Sync { fn get(&self, content_id: &Cid) -> Result>>; fn put(&self, node: &Node) -> Result<()>; fn delete(&self, content_id: &Cid) -> Result<()>; @@ -17,7 +17,7 @@ pub trait NodeStorage { /// [`NodeStorage`] implementation backed by a shared LevelDB instance. pub struct LeveldbNodeStorage { - shared: Rc, + shared: Arc, _marker: std::marker::PhantomData<(P, M)>, } @@ -38,7 +38,7 @@ impl LeveldbNodeStorage { } /// Creates the storage from an existing [`SharedLeveldb`] handle. - pub fn new(shared: Rc) -> Self { + pub fn new(shared: Arc) -> Self { Self { shared, _marker: std::marker::PhantomData, @@ -62,7 +62,6 @@ impl LeveldbNodeStorage { { self.shared .db() - .borrow_mut() .put(key, value) .map_err(GraphError::Storage)?; } @@ -78,7 +77,6 @@ impl LeveldbNodeStorage { { self.shared .db() - .borrow_mut() .delete(key) .map_err(GraphError::Storage)?; } @@ -87,7 +85,7 @@ impl LeveldbNodeStorage { } impl SharedLeveldbAccess for LeveldbNodeStorage { - fn shared_leveldb(&self) -> Option> { + fn shared_leveldb(&self) -> Option> { Some(self.shared.clone()) } } @@ -99,7 +97,7 @@ where { fn get(&self, cid: &Cid) -> Result>> { let key = Self::make_key(cid); - match self.shared.db().borrow_mut().get(&key) { + match self.shared.db().get(&key) { Some(raw) => { let node = Node::from_bytes(&raw).map_err(|e| GraphError::NodeOperation(e.to_string()))?; @@ -131,7 +129,6 @@ where let mut iter = self .shared .db() - .borrow_mut() .new_iter() .map_err(GraphError::Storage)?; iter.seek_to_first(); diff --git a/src/repo.rs b/src/repo.rs index 807dc0b..6f1dfce 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -18,7 +18,7 @@ use cid::Cid; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::fmt::Debug; -use std::rc::Rc; +use std::sync::Arc; struct PendingNode { cid: Cid, @@ -151,7 +151,7 @@ where Ok(path) } - fn shared_leveldb(&self) -> Result> { + 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()) })?; @@ -160,7 +160,7 @@ where CrdtError::Internal("node storage does not support batching".into()) })?; - if !Rc::ptr_eq(&op_db, &node_db) { + if !Arc::ptr_eq(&op_db, &node_db) { return Err(CrdtError::Internal( "operation and node storage must share the same LevelDB instance for transactions" .into(), @@ -221,6 +221,9 @@ where "a transaction is already active on the shared LevelDB".to_string(), ), BatchError::Commit(status) => CrdtError::Storage(status), + BatchError::LockPoisoned => CrdtError::Internal( + "shared LevelDB lock was poisoned".to_string(), + ), }) } @@ -498,7 +501,7 @@ mod tests { use crate::graph::error::GraphError; use crate::graph::storage::LeveldbNodeStorage; use rusty_leveldb::{Status, StatusCode}; - use std::cell::Cell; + use std::sync::atomic::{AtomicBool, Ordering}; use tempfile::tempdir; use ulid::Ulid; @@ -536,35 +539,37 @@ mod tests { struct FailingOperationStorage { inner: S, - fail_next: Cell, + fail_next: AtomicBool, } impl FailingOperationStorage { fn new(inner: S) -> Self { Self { inner, - fail_next: Cell::new(false), + fail_next: AtomicBool::new(false), } } fn fail_on_first(inner: S) -> Self { Self { inner, - fail_next: Cell::new(true), + fail_next: AtomicBool::new(true), } } fn fail_on_next(&self) { - self.fail_next.set(true); + self.fail_next.store(true, Ordering::SeqCst); } } impl OperationStorage for FailingOperationStorage where S: OperationStorage, + ContentId: Send + Sync, + T: Send + Sync, { fn save_operation(&self, op: &Operation) -> crate::crdt::error::Result<()> { - if self.fail_next.replace(false) { + if self.fail_next.swap(false, Ordering::SeqCst) { Err(CrdtError::Internal( "forced failure for testing".to_string(), )) @@ -596,21 +601,21 @@ mod tests { where S: SharedLeveldbAccess, { - fn shared_leveldb(&self) -> Option> { + fn shared_leveldb(&self) -> Option> { self.inner.shared_leveldb() } } struct FailingNodeStorage { inner: S, - fail_next_put: Cell, + fail_next_put: AtomicBool, } impl FailingNodeStorage { fn fail_on_first_put(inner: S) -> Self { Self { inner, - fail_next_put: Cell::new(true), + fail_next_put: AtomicBool::new(true), } } } @@ -618,13 +623,15 @@ mod tests { impl NodeStorage for FailingNodeStorage where S: NodeStorage, + P: Send + Sync, + M: Send + Sync, { 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) { + if self.fail_next_put.swap(false, Ordering::SeqCst) { Err(GraphError::Internal( "injected node storage failure".to_string(), )) @@ -646,7 +653,7 @@ mod tests { where S: SharedLeveldbAccess, { - fn shared_leveldb(&self) -> Option> { + fn shared_leveldb(&self) -> Option> { self.inner.shared_leveldb() } } diff --git a/src/storage/shared_leveldb.rs b/src/storage/shared_leveldb.rs index 12b112f..7dc78aa 100644 --- a/src/storage/shared_leveldb.rs +++ b/src/storage/shared_leveldb.rs @@ -1,39 +1,42 @@ use rusty_leveldb::{Options, Status, WriteBatch, DB as Database}; -use std::cell::RefCell; use std::path::Path; -use std::rc::Rc; +use std::sync::{Arc, Mutex, MutexGuard}; #[derive(Debug)] pub enum BatchError { Unsupported, AlreadyActive, Commit(Status), + LockPoisoned, } pub struct SharedLeveldb { - db: RefCell, - active_batch: RefCell>, + db: Mutex, + active_batch: Mutex>, #[cfg(test)] - commit_fail_status: RefCell>, + commit_fail_status: Mutex>, } impl SharedLeveldb { - pub fn open>(path: P) -> Result, Status> { + 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), + Ok(Arc::new(Self { + db: Mutex::new(db), + active_batch: Mutex::new(None), #[cfg(test)] - commit_fail_status: RefCell::new(None), + commit_fail_status: Mutex::new(None), })) } pub fn begin_batch(&self) -> Result, BatchError> { - let mut slot = self.active_batch.borrow_mut(); + let mut slot = self + .active_batch + .lock() + .map_err(|_| BatchError::LockPoisoned)?; if slot.is_some() { return Err(BatchError::AlreadyActive); } @@ -45,31 +48,47 @@ impl SharedLeveldb { } fn commit_batch(&self) -> Result<(), Status> { - let mut slot = self.active_batch.borrow_mut(); + let mut slot = self.active_batch.lock().map_err(|_| { + Status::new(rusty_leveldb::StatusCode::LockError, "Lock poisoned") + })?; let Some(batch) = slot.take() else { return Ok(()); }; #[cfg(test)] - if let Some(status) = self.commit_fail_status.borrow_mut().take() { + if let Some(status) = self + .commit_fail_status + .lock() + .ok() + .and_then(|mut s| s.take()) + { return Err(status); } - self.db.borrow_mut().write(batch, true) + self.db + .lock() + .map_err(|_| Status::new(rusty_leveldb::StatusCode::LockError, "Lock poisoned"))? + .write(batch, true) } fn abort_batch(&self) { - self.active_batch.borrow_mut().take(); + if let Ok(mut slot) = self.active_batch.lock() { + slot.take(); + } } pub fn with_active_batch(&self, f: F) -> Option where F: FnOnce(&mut WriteBatch) -> R, { - let mut slot = self.active_batch.borrow_mut(); + let mut slot = self.active_batch.lock().ok()?; slot.as_mut().map(f) } - pub fn db(&self) -> &RefCell { - &self.db + pub fn db(&self) -> MutexGuard<'_, Database> { + self.db.lock().expect("Database lock poisoned") + } + + pub fn try_db(&self) -> Result, BatchError> { + self.db.lock().map_err(|_| BatchError::LockPoisoned) } } @@ -95,13 +114,15 @@ impl Drop for LeveldbBatchGuard<'_> { } pub trait SharedLeveldbAccess { - fn shared_leveldb(&self) -> Option>; + 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); + if let Ok(mut slot) = self.commit_fail_status.lock() { + slot.replace(status); + } } } @@ -146,7 +167,6 @@ mod tests { let stored = shared .db() - .borrow_mut() .get(key) .expect("value should exist after commit"); assert_eq!(stored.as_slice(), value); @@ -166,7 +186,7 @@ mod tests { // guard dropped here without commit } - let result = shared.db().borrow_mut().get(key); + let result = shared.db().get(key); assert!( result.is_none(), "value should not be persisted when batch guard is dropped without commit" From 5ff3f253a49038c28f5abe960a54cb16f6b62ede Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 30 Nov 2025 05:30:56 +0900 Subject: [PATCH 3/4] Refactor code for improved readability --- src/crdt/storage.rs | 14 +++++++------- src/graph/dag.rs | 35 ++++++++++++++++++++++++++++++----- src/graph/storage.rs | 11 ++--------- src/repo.rs | 6 +++--- src/storage/shared_leveldb.rs | 7 ++++--- 5 files changed, 46 insertions(+), 27 deletions(-) diff --git a/src/crdt/storage.rs b/src/crdt/storage.rs index 1877ba5..9cb2480 100644 --- a/src/crdt/storage.rs +++ b/src/crdt/storage.rs @@ -90,8 +90,12 @@ impl SharedLeveldbAccess for LeveldbStorage { impl OperationStorage for LeveldbStorage where - ContentId: - serde::Serialize + for<'de> serde::Deserialize<'de> + PartialEq + std::fmt::Debug + Send + Sync, + ContentId: serde::Serialize + + for<'de> serde::Deserialize<'de> + + PartialEq + + std::fmt::Debug + + Send + + Sync, T: serde::Serialize + for<'de> serde::Deserialize<'de> + std::fmt::Debug + Send + Sync, { fn begin_batch(&self) -> std::result::Result, BatchError> { @@ -106,11 +110,7 @@ where fn load_operations(&self, genesis: &ContentId) -> Result>> { let mut result = Vec::new(); - let mut iter = self - .shared - .db() - .new_iter() - .map_err(CrdtError::Storage)?; + let mut iter = self.shared.db().new_iter().map_err(CrdtError::Storage)?; iter.seek_to_first(); let mut key = Vec::new(); diff --git a/src/graph/dag.rs b/src/graph/dag.rs index 7fdbe00..eed4b2f 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -491,7 +491,12 @@ mod tests { None => return Ok(None), }; - let ts = *self.timestamps.lock().unwrap().get(content_id).unwrap_or(&0); + let ts = *self + .timestamps + .lock() + .unwrap() + .get(content_id) + .unwrap_or(&0); fn find_genesis(edges: &HashMap>, cid: &Cid) -> Cid { let mut current = *cid; @@ -862,7 +867,12 @@ mod tests { fn test_get_genesis_from_genesis_node() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.lock().unwrap().entry(genesis_cid).or_default(); + storage + .edges + .lock() + .unwrap() + .entry(genesis_cid) + .or_default(); let dag = DagGraph::>::new(storage); let result = dag.get_genesis(&genesis_cid); @@ -897,7 +907,12 @@ mod tests { fn test_calculate_latest_genesis_only() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.lock().unwrap().entry(genesis_cid).or_default(); + storage + .edges + .lock() + .unwrap() + .entry(genesis_cid) + .or_default(); let dag = DagGraph::>::new(storage); let result = dag.calculate_latest(&genesis_cid).unwrap(); assert_eq!(result, Some(genesis_cid)); @@ -942,7 +957,12 @@ mod tests { fn test_get_nodes_by_genesis_genesis_only() { let storage = MockStorage::new(); let genesis_cid = create_test_content_id(b"genesis"); - storage.edges.lock().unwrap().entry(genesis_cid).or_default(); + storage + .edges + .lock() + .unwrap() + .entry(genesis_cid) + .or_default(); let dag = DagGraph::>::new(storage); let result = dag.get_nodes_by_genesis(&genesis_cid).unwrap(); assert_eq!(result, vec![genesis_cid]); @@ -970,7 +990,12 @@ mod tests { let v1_cid = create_test_content_id(b"v1"); let unrelated_cid = create_test_content_id(b"unrelated"); storage.setup_graph(&[(genesis1_cid, v1_cid)]); - storage.edges.lock().unwrap().entry(unrelated_cid).or_default(); + storage + .edges + .lock() + .unwrap() + .entry(unrelated_cid) + .or_default(); let dag = DagGraph::>::new(storage); let mut result = dag.get_nodes_by_genesis(&genesis1_cid).unwrap(); result.sort(); diff --git a/src/graph/storage.rs b/src/graph/storage.rs index 6f511aa..f0be87a 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -75,10 +75,7 @@ impl LeveldbNodeStorage { .with_active_batch(|batch| batch.delete(key)) .is_none() { - self.shared - .db() - .delete(key) - .map_err(GraphError::Storage)?; + self.shared.db().delete(key).map_err(GraphError::Storage)?; } Ok(()) } @@ -126,11 +123,7 @@ where /// 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 - .shared - .db() - .new_iter() - .map_err(GraphError::Storage)?; + let mut iter = self.shared.db().new_iter().map_err(GraphError::Storage)?; iter.seek_to_first(); let mut key = Vec::new(); let mut value = Vec::new(); diff --git a/src/repo.rs b/src/repo.rs index 6f1dfce..67b1557 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -221,9 +221,9 @@ where "a transaction is already active on the shared LevelDB".to_string(), ), BatchError::Commit(status) => CrdtError::Storage(status), - BatchError::LockPoisoned => CrdtError::Internal( - "shared LevelDB lock was poisoned".to_string(), - ), + BatchError::LockPoisoned => { + CrdtError::Internal("shared LevelDB lock was poisoned".to_string()) + } }) } diff --git a/src/storage/shared_leveldb.rs b/src/storage/shared_leveldb.rs index 7dc78aa..ec6d9a8 100644 --- a/src/storage/shared_leveldb.rs +++ b/src/storage/shared_leveldb.rs @@ -48,9 +48,10 @@ impl SharedLeveldb { } fn commit_batch(&self) -> Result<(), Status> { - let mut slot = self.active_batch.lock().map_err(|_| { - Status::new(rusty_leveldb::StatusCode::LockError, "Lock poisoned") - })?; + let mut slot = self + .active_batch + .lock() + .map_err(|_| Status::new(rusty_leveldb::StatusCode::LockError, "Lock poisoned"))?; let Some(batch) = slot.take() else { return Ok(()); }; From 2ad1d7dd1b8280b221aebc9e6b882aec9120412d Mon Sep 17 00:00:00 2001 From: Yu-da-1 Date: Tue, 2 Dec 2025 00:46:16 +0900 Subject: [PATCH 4/4] fix: update use enum for content policy type and switch DAG timestamps to nanoseconds --- src/convergence/metadata.rs | 42 +++++++++++++++++++++++++++++++------ src/convergence/resolver.rs | 2 +- src/graph/dag.rs | 3 ++- 3 files changed, 39 insertions(+), 8 deletions(-) diff --git a/src/convergence/metadata.rs b/src/convergence/metadata.rs index 047792e..8fbd514 100644 --- a/src/convergence/metadata.rs +++ b/src/convergence/metadata.rs @@ -1,11 +1,36 @@ use serde::{Deserialize, Serialize}; +/// Built-in and custom convergence policy types. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub enum PolicyType { + /// Last-Write-Wins policy. + Lww, + /// Any non-builtin policy, identified by its name. + Custom(String), +} + +impl From<&str> for PolicyType { + fn from(value: &str) -> Self { + match value { + "lww" => PolicyType::Lww, + other => PolicyType::Custom(other.to_string()), + } + } +} + +impl From for PolicyType { + fn from(value: String) -> Self { + PolicyType::from(value.as_str()) + } +} + /// Metadata that stores information required for convergence policies. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct ContentMetadata { - /// Policy type name (e.g. "lww", "text", "custom-policy"). - /// When this is `None`, it falls back to the default policy (currently "lww"). - policy_type: Option, + /// Policy type (e.g. Lww, custom named policy). + /// + /// When this is `None`, it falls back to the default policy (currently Lww). + policy_type: Option, } impl ContentMetadata { @@ -15,15 +40,20 @@ impl ContentMetadata { } /// Create metadata that uses the specified policy. - pub fn with_policy(policy_type: impl Into) -> Self { + /// + /// This accepts either a concrete `PolicyType` or a string like `"lww"` or `"custom-policy"`. + pub fn with_policy(policy_type: impl Into) -> Self { Self { policy_type: Some(policy_type.into()), } } - /// Return the configured policy type; falls back to "lww" when unspecified. + /// Return the configured policy type name; falls back to `"lww"` when unspecified. pub fn policy_type(&self) -> &str { - self.policy_type.as_deref().unwrap_or("lww") + match &self.policy_type { + Some(PolicyType::Lww) | None => "lww", + Some(PolicyType::Custom(name)) => name.as_str(), + } } } diff --git a/src/convergence/resolver.rs b/src/convergence/resolver.rs index fd58424..bd98240 100644 --- a/src/convergence/resolver.rs +++ b/src/convergence/resolver.rs @@ -101,7 +101,7 @@ where SystemTime::now() .duration_since(UNIX_EPOCH) .map_err(|e| CrdtError::Internal(format!("timestamp error: {e}"))) - .map(|duration| duration.as_secs()) + .map(|duration| duration.as_nanos() as u64) } } diff --git a/src/graph/dag.rs b/src/graph/dag.rs index eed4b2f..d27f6ca 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -189,11 +189,12 @@ where Ok(result) } + /// Returns the current time in nanoseconds since the Unix epoch. fn current_timestamp() -> Result { SystemTime::now() .duration_since(UNIX_EPOCH) .map_err(GraphError::Timestamp) - .map(|d| d.as_secs()) + .map(|d| d.as_nanos() as u64) } /// Check if adding an edge (new node with parents) would create a cycle