From d01e797d90e44b44be6333cbbb65cb3b6f8332df Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 14:20:00 +0900 Subject: [PATCH 1/7] Update version to 0.2.0 and add import_operation logic --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/graph/dag.rs | 50 ++++++++ src/repo.rs | 313 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 365 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4ac928d..96eb801 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -250,7 +250,7 @@ checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" [[package]] name = "crsl-lib" -version = "0.1.0" +version = "0.2.0" dependencies = [ "bincode", "cid", diff --git a/Cargo.toml b/Cargo.toml index 43cddc8..f1dbdc8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "crsl-lib" -version = "0.1.0" +version = "0.2.0" edition = "2021" rust-version = "1.79" diff --git a/src/graph/dag.rs b/src/graph/dag.rs index d27f6ca..9136dd9 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -88,6 +88,29 @@ where pub fn prepare_genesis_node(&mut self, payload: P, metadata: M) -> Result<(Cid, Node)> { let timestamp = Self::current_timestamp()?; + self.prepare_genesis_node_with_timestamp(payload, timestamp, metadata) + } + + /// Creates a genesis node with a specified timestamp (for replication). + /// + /// This method is used when importing operations from other replicas, + /// where the original timestamp must be preserved to maintain CID consistency. + /// + /// # Arguments + /// + /// * `payload` - The payload data for the node + /// * `timestamp` - The timestamp to use (from the original node) + /// * `metadata` - The metadata for the node + /// + /// # Returns + /// + /// A tuple of (CID, Node) for the created genesis node + pub fn prepare_genesis_node_with_timestamp( + &mut self, + payload: P, + timestamp: u64, + metadata: M, + ) -> Result<(Cid, Node)> { let node = Node::new_genesis(payload, timestamp, metadata); let cid = node.content_id()?; Ok((cid, node)) @@ -101,6 +124,33 @@ where metadata: M, ) -> Result<(Cid, Node)> { let timestamp = Self::current_timestamp()?; + self.prepare_child_node_with_timestamp(payload, parents, genesis, timestamp, metadata) + } + + /// Creates a child node with a specified timestamp (for replication). + /// + /// This method is used when importing operations from other replicas, + /// where the original timestamp must be preserved to maintain CID consistency. + /// + /// # Arguments + /// + /// * `payload` - The payload data for the node + /// * `parents` - The parent CIDs for this node + /// * `genesis` - The genesis CID this node belongs to + /// * `timestamp` - The timestamp to use (from the original node) + /// * `metadata` - The metadata for the node + /// + /// # Returns + /// + /// A tuple of (CID, Node) for the created child node + pub fn prepare_child_node_with_timestamp( + &mut self, + payload: P, + parents: Vec, + genesis: Cid, + timestamp: u64, + metadata: M, + ) -> Result<(Cid, Node)> { let node = Node::new_child(payload, parents.clone(), genesis, timestamp, metadata); let cid = node.content_id()?; diff --git a/src/repo.rs b/src/repo.rs index 67b1557..ce4a1bf 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -65,6 +65,68 @@ where self.commit_operation_internal(op, false) } + /// Imports an operation from another replica with the original node timestamp. + /// + /// Unlike `commit_operation`, this method preserves the original timestamp + /// from the source replica, ensuring that the resulting CID matches the + /// original. This is essential for CRDT replication where CID consistency + /// across replicas is required. + /// + /// # Arguments + /// + /// * `op` - The operation to import (with genesis already set for Create operations) + /// * `node_timestamp` - The original DAG node timestamp from the source replica + /// + /// # Returns + /// + /// The CID of the imported node (should match the original CID) + /// + /// # Errors + /// + /// Returns an error if the operation cannot be applied or if there are + /// consistency issues with the DAG structure. + pub fn import_operation( + &mut self, + op: Operation, + node_timestamp: u64, + ) -> Result { + let shared = self.shared_leveldb()?; + let batch_guard = Self::begin_shared_batch(&shared)?; + let mut pending_nodes: Vec = Vec::new(); + + let cid = match op.kind.clone() { + OperationType::Create(payload) => { + self.stage_create_with_timestamp(payload, &op, node_timestamp, &mut pending_nodes)? + } + OperationType::Update(payload) => { + self.stage_update_with_timestamp( + payload, + &op, + node_timestamp, + &mut pending_nodes, + )? + } + OperationType::Delete => { + self.stage_delete_with_timestamp(&op, node_timestamp, &mut pending_nodes)? + } + OperationType::Merge(payload) => { + self.stage_merge_with_timestamp(payload, &op, node_timestamp, &mut pending_nodes)? + } + }; + + 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() } @@ -281,6 +343,30 @@ where Ok(cid) } + /// Stages a Create operation with a specified timestamp (for replication). + /// Unlike `stage_create`, this preserves the original genesis CID from the operation. + fn stage_create_with_timestamp( + &mut self, + payload: Payload, + op: &Operation, + timestamp: u64, + pending_nodes: &mut Vec, + ) -> Result { + let (genesis_cid, node) = self + .dag + .prepare_genesis_node_with_timestamp(payload, timestamp, ContentMetadata::default())?; + + // Verify that the computed CID matches the expected genesis from the operation + if genesis_cid != op.genesis { + return Err(CrdtError::Internal(format!( + "CID mismatch during import: expected {}, got {}", + op.genesis, genesis_cid + ))); + } + + self.stage_prepared_node(genesis_cid, node, pending_nodes) + } + fn stage_update( &mut self, payload: Payload, @@ -295,6 +381,26 @@ where self.stage_prepared_node(cid, node, pending_nodes) } + /// Stages an Update operation with a specified timestamp (for replication). + fn stage_update_with_timestamp( + &mut self, + payload: Payload, + op: &Operation, + timestamp: u64, + pending_nodes: &mut Vec, + ) -> Result { + let metadata = + self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + let (cid, node) = self.dag.prepare_child_node_with_timestamp( + payload, + op.parents.clone(), + op.genesis, + timestamp, + metadata, + )?; + self.stage_prepared_node(cid, node, pending_nodes) + } + fn stage_delete( &mut self, op: &Operation, @@ -326,6 +432,63 @@ where self.stage_prepared_node(cid, node, pending_nodes) } + /// Stages a Delete operation with a specified timestamp (for replication). + fn stage_delete_with_timestamp( + &mut self, + op: &Operation, + timestamp: u64, + 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(|(ts, _)| *ts) + .map(|(_, payload)| payload) + .ok_or_else(|| { + CrdtError::Internal(format!( + "content must exist for delete operation: {}", + op.genesis + )) + })?; + + let metadata = + self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + let (cid, node) = self.dag.prepare_child_node_with_timestamp( + last_payload, + op.parents.clone(), + op.genesis, + timestamp, + metadata, + )?; + self.stage_prepared_node(cid, node, pending_nodes) + } + + /// Stages a Merge operation with a specified timestamp (for replication). + fn stage_merge_with_timestamp( + &mut self, + payload: Payload, + op: &Operation, + timestamp: u64, + pending_nodes: &mut Vec, + ) -> Result { + let metadata = + self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + let (cid, node) = self.dag.prepare_child_node_with_timestamp( + payload, + op.parents.clone(), + op.genesis, + timestamp, + metadata, + )?; + self.stage_prepared_node(cid, node, pending_nodes) + } + fn stage_prepared_node( &mut self, cid: Cid, @@ -482,6 +645,37 @@ where Ok(genesis_node.metadata().clone()) } } + + /// Resolves metadata for import operations. + /// This is more lenient than `resolve_metadata_for_commit` as parent nodes + /// may not exist yet during replication (operations may arrive out of order). + fn resolve_metadata_for_import( + &self, + genesis: &Cid, + parents: &[Cid], + pending_nodes: &[PendingNode], + ) -> Result { + // Try to get metadata from parents first + if let Some(parent) = parents.first() { + if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == parent) { + return Ok(pending.metadata.clone()); + } + if let Ok(Some(node)) = self.dag.get_node(parent) { + return Ok(node.metadata().clone()); + } + } + + // Try to get metadata from genesis + if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == genesis) { + return Ok(pending.metadata.clone()); + } + if let Ok(Some(genesis_node)) = self.dag.get_node(genesis) { + return Ok(genesis_node.metadata().clone()); + } + + // Default metadata if nothing is found (for out-of-order imports) + Ok(ContentMetadata::default()) + } fn node_characteristics(&self, cid: &Cid) -> Result<(bool, u64)> { let node = self .dag @@ -1508,4 +1702,123 @@ mod tests { panic!("branch or merge node missing from linear history"); } } + + #[test] + fn test_import_operation_preserves_cid() { + let (mut repo1, _dir1) = setup_test_repo(); + let (mut repo2, _dir2) = setup_test_repo(); + + // Create content in repo1 + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"import-test").unwrap(), + ); + let payload = TestPayload("test content".to_string()); + let op = make_test_operation(initial_genesis, OperationType::Create(payload.clone())); + + let cid1 = repo1.commit_operation(op.clone()).unwrap(); + + // Get the node timestamp from repo1 + let node = repo1.dag.get_node(&cid1).unwrap().unwrap(); + let node_timestamp = node.timestamp(); + + // Create the operation with the correct genesis CID for import + let mut import_op = make_test_operation(cid1, OperationType::Create(payload)); + import_op.genesis = cid1; + + // Import the operation into repo2 with the original timestamp + let cid2 = repo2.import_operation(import_op, node_timestamp).unwrap(); + + // CIDs should match + assert_eq!(cid1, cid2, "CIDs should be identical after import"); + + // Verify the content can be retrieved using the original CID + assert!( + repo2.latest(&cid1).is_some(), + "Should be able to get latest using original CID" + ); + assert_eq!(repo2.latest(&cid1).unwrap(), cid1); + } + + #[test] + fn test_import_operation_update_preserves_cid() { + let (mut repo1, _dir1) = setup_test_repo(); + let (mut repo2, _dir2) = setup_test_repo(); + + // Create initial content in repo1 + let initial_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"import-update-test").unwrap(), + ); + let create_payload = TestPayload("initial".to_string()); + let create_op = + make_test_operation(initial_genesis, OperationType::Create(create_payload.clone())); + let genesis_cid = repo1.commit_operation(create_op.clone()).unwrap(); + + // Get genesis node timestamp + let genesis_node = repo1.dag.get_node(&genesis_cid).unwrap().unwrap(); + let genesis_timestamp = genesis_node.timestamp(); + + // Import genesis into repo2 + let mut import_create_op = + make_test_operation(genesis_cid, OperationType::Create(create_payload)); + import_create_op.genesis = genesis_cid; + let imported_genesis = repo2 + .import_operation(import_create_op, genesis_timestamp) + .unwrap(); + assert_eq!(genesis_cid, imported_genesis); + + // Create update in repo1 + sleep_for_ordering(); + let update_payload = TestPayload("updated".to_string()); + let update_op = + make_test_operation(genesis_cid, OperationType::Update(update_payload.clone())); + let update_cid = repo1.commit_operation(update_op).unwrap(); + + // Get update node info from repo1 + let update_node = repo1.dag.get_node(&update_cid).unwrap().unwrap(); + let update_timestamp = update_node.timestamp(); + let update_parents = update_node.parents().clone(); + + // Import update into repo2 + let mut import_update_op = + make_test_operation(genesis_cid, OperationType::Update(update_payload)); + import_update_op.parents = update_parents; + let imported_update = repo2 + .import_operation(import_update_op, update_timestamp) + .unwrap(); + + // CIDs should match + assert_eq!( + update_cid, imported_update, + "Update CIDs should be identical after import" + ); + + // Verify latest points to the update + assert_eq!(repo2.latest(&genesis_cid).unwrap(), update_cid); + } + + #[test] + fn test_import_operation_rejects_cid_mismatch() { + let (mut repo, _dir) = setup_test_repo(); + + // Create an operation with a genesis CID that won't match the computed CID + let wrong_genesis = Cid::new_v1( + 0x55, + multihash::Multihash::<64>::wrap(0x12, b"wrong-genesis").unwrap(), + ); + let payload = TestPayload("test content".to_string()); + let mut op = make_test_operation(wrong_genesis, OperationType::Create(payload)); + op.genesis = wrong_genesis; // This won't match the computed CID + + // Import should fail due to CID mismatch + let result = repo.import_operation(op, 12345); + assert!(result.is_err()); + match result { + Err(CrdtError::Internal(msg)) => { + assert!(msg.contains("CID mismatch")); + } + other => panic!("Expected CID mismatch error, got: {:?}", other), + } + } } From cdd1c798b72b5738a1ee080c35f306f85b41d3fd Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 14:46:00 +0900 Subject: [PATCH 2/7] Create merge node with timestamp, add nodes with timestamp --- src/convergence/resolver.rs | 34 +++--- src/crdt/operation.rs | 40 +++++++ src/crdt/reducer.rs | 2 + src/graph/dag.rs | 110 ++++++++++--------- src/repo.rs | 213 ++++++++++++++++++++---------------- 5 files changed, 237 insertions(+), 162 deletions(-) diff --git a/src/convergence/resolver.rs b/src/convergence/resolver.rs index bd98240..2fadac2 100644 --- a/src/convergence/resolver.rs +++ b/src/convergence/resolver.rs @@ -24,11 +24,21 @@ where } } + /// Creates a merge node from the given heads. + /// + /// # Arguments + /// + /// * `heads` - The head CIDs to merge + /// * `dag` - The DAG graph + /// * `genesis` - The genesis CID + /// * `timestamp` - The timestamp to use for the merge node + /// * `policy` - The merge policy to use pub fn create_merge_node( &self, heads: &[Cid], dag: &DagGraph, genesis: Cid, + timestamp: u64, policy: &dyn MergePolicy

, ) -> CrdtResult> where @@ -45,7 +55,6 @@ where let inputs = self.collect_inputs(heads, dag)?; let merged_payload = policy.resolve(&inputs); let metadata = self.merge_metadata(heads, dag)?; - let timestamp = Self::current_timestamp()?; Ok(Node::new_child( merged_payload, heads.to_vec(), @@ -95,14 +104,6 @@ where .ok_or_else(|| CrdtError::Internal(format!("Head node not found: {first_head}")))?; Ok(node.metadata().clone()) } - - fn current_timestamp() -> CrdtResult { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_err(|e| CrdtError::Internal(format!("timestamp error: {e}"))) - .map(|duration| duration.as_nanos() as u64) - } } #[cfg(test)] @@ -231,15 +232,22 @@ mod tests { }; let resolver = ConflictResolver::::new(); + let merge_timestamp = 100; let merge_node = resolver - .create_merge_node(&[head_a_cid, head_b_cid], &dag, genesis_cid, &policy) + .create_merge_node( + &[head_a_cid, head_b_cid], + &dag, + genesis_cid, + merge_timestamp, + &policy, + ) .unwrap(); assert_eq!(merge_node.payload(), "merged"); assert_eq!(merge_node.parents(), &vec![head_a_cid, head_b_cid]); assert_eq!(merge_node.metadata(), &metadata); assert_eq!(merge_node.genesis, Some(genesis_cid)); - assert!(merge_node.timestamp() > 0); + assert_eq!(merge_node.timestamp(), merge_timestamp); } #[test] @@ -249,7 +257,7 @@ mod tests { let policy = LwwMergePolicy; let genesis = create_test_cid("genesis"); - let result = resolver.create_merge_node(&[], &dag, genesis, &policy); + let result = resolver.create_merge_node(&[], &dag, genesis, 100, &policy); assert!(matches!( result, @@ -265,7 +273,7 @@ mod tests { let missing_head = create_test_cid("missing-head"); let genesis = create_test_cid("genesis"); - let result = resolver.create_merge_node(&[missing_head], &dag, genesis, &policy); + let result = resolver.create_merge_node(&[missing_head], &dag, genesis, 100, &policy); assert!(matches!( result, diff --git a/src/crdt/operation.rs b/src/crdt/operation.rs index 4ec622e..756c040 100644 --- a/src/crdt/operation.rs +++ b/src/crdt/operation.rs @@ -61,6 +61,11 @@ pub struct Operation { pub author: Author, #[serde(default = "Vec::new")] pub parents: Vec, + /// Optional timestamp for the DAG node (used for replication). + /// When set, this timestamp is used for CID generation instead of the current time. + /// This ensures CID consistency across replicas. + #[serde(default)] + pub node_timestamp: Option, } impl Operation @@ -89,6 +94,41 @@ where timestamp, author, parents: Vec::new(), + node_timestamp: None, + } + } + + /// Creates a new operation with a specified node timestamp (for replication). + /// + /// When importing operations from other replicas, use this constructor + /// to preserve the original node timestamp for CID consistency. + /// + /// # Arguments + /// + /// * `genesis` - ID of the content being operated on + /// * `kind` - Type of operation and its payload + /// * `author` - User/system performing the operation + /// * `node_timestamp` - The original DAG node timestamp from the source replica + /// + /// # Returns + /// + /// A newly created operation object with the specified node timestamp + pub fn with_node_timestamp( + genesis: ContentId, + kind: OperationType, + author: Author, + node_timestamp: u64, + ) -> Self { + let timestamp = next_monotonic_timestamp(); + let id = Ulid::new(); + Self { + id, + genesis, + kind, + timestamp, + author, + parents: Vec::new(), + node_timestamp: Some(node_timestamp), } } diff --git a/src/crdt/reducer.rs b/src/crdt/reducer.rs index 24403cb..a9444b8 100644 --- a/src/crdt/reducer.rs +++ b/src/crdt/reducer.rs @@ -52,6 +52,7 @@ mod tests { timestamp: ts, author: "test".into(), parents: Vec::new(), + node_timestamp: None, } } @@ -69,6 +70,7 @@ mod tests { timestamp: ts, author: "test".into(), parents: Vec::new(), + node_timestamp: None, } } diff --git a/src/graph/dag.rs b/src/graph/dag.rs index 9136dd9..c30bffb 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -4,7 +4,6 @@ use crate::graph::storage::NodeStorage; use cid::Cid; use std::collections::{HashMap, HashSet}; use std::marker::PhantomData; -use std::time::{SystemTime, UNIX_EPOCH}; /// Directed Acyclic Graph(DAG) Structure /// @@ -39,9 +38,25 @@ where } } - pub fn add_node(&mut self, payload: P, parents: Vec, metadata: M) -> Result { + /// Adds a node to the DAG with a specified timestamp. + /// + /// If parents is empty, creates a genesis node. Otherwise, creates a child node. + /// + /// # Arguments + /// + /// * `payload` - The payload data for the node + /// * `parents` - Parent CIDs (empty for genesis nodes) + /// * `timestamp` - The timestamp for CID generation + /// * `metadata` - The metadata for the node + pub fn add_node( + &mut self, + payload: P, + parents: Vec, + timestamp: u64, + metadata: M, + ) -> Result { if parents.is_empty() { - let (cid, node) = self.prepare_genesis_node(payload, metadata)?; + let (cid, node) = self.prepare_genesis_node(payload, timestamp, metadata)?; return self.persist_and_cache(cid, node); } @@ -66,46 +81,55 @@ where GraphError::Internal("child node requires at least one parent".to_string()) })?; - let (cid, node) = self.prepare_child_node(payload, parents, genesis, metadata)?; + let (cid, node) = self.prepare_child_node(payload, parents, genesis, timestamp, metadata)?; self.persist_and_cache(cid, node) } - pub fn add_genesis_node(&mut self, payload: P, metadata: M) -> Result { - let (cid, node) = self.prepare_genesis_node(payload, metadata)?; + /// Adds a genesis node to the DAG with a specified timestamp. + /// + /// # Arguments + /// + /// * `payload` - The payload data for the node + /// * `timestamp` - The timestamp for CID generation + /// * `metadata` - The metadata for the node + pub fn add_genesis_node(&mut self, payload: P, timestamp: u64, metadata: M) -> Result { + let (cid, node) = self.prepare_genesis_node(payload, timestamp, metadata)?; self.persist_and_cache(cid, node) } + /// Adds a child node to the DAG with a specified timestamp. + /// + /// # Arguments + /// + /// * `payload` - The payload data for the node + /// * `parents` - The parent CIDs for this node + /// * `genesis` - The genesis CID this node belongs to + /// * `timestamp` - The timestamp for CID generation + /// * `metadata` - The metadata for the node pub fn add_child_node( &mut self, payload: P, parents: Vec, genesis: Cid, + timestamp: u64, metadata: M, ) -> Result { - let (cid, node) = self.prepare_child_node(payload, parents, genesis, metadata)?; + let (cid, node) = self.prepare_child_node(payload, parents, genesis, timestamp, 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()?; - self.prepare_genesis_node_with_timestamp(payload, timestamp, metadata) - } - - /// Creates a genesis node with a specified timestamp (for replication). - /// - /// This method is used when importing operations from other replicas, - /// where the original timestamp must be preserved to maintain CID consistency. + /// Prepares a genesis node with a specified timestamp. /// /// # Arguments /// /// * `payload` - The payload data for the node - /// * `timestamp` - The timestamp to use (from the original node) + /// * `timestamp` - The timestamp for CID generation /// * `metadata` - The metadata for the node /// /// # Returns /// /// A tuple of (CID, Node) for the created genesis node - pub fn prepare_genesis_node_with_timestamp( + pub fn prepare_genesis_node( &mut self, payload: P, timestamp: u64, @@ -116,34 +140,20 @@ where Ok((cid, node)) } - pub fn prepare_child_node( - &mut self, - payload: P, - parents: Vec, - genesis: Cid, - metadata: M, - ) -> Result<(Cid, Node)> { - let timestamp = Self::current_timestamp()?; - self.prepare_child_node_with_timestamp(payload, parents, genesis, timestamp, metadata) - } - - /// Creates a child node with a specified timestamp (for replication). - /// - /// This method is used when importing operations from other replicas, - /// where the original timestamp must be preserved to maintain CID consistency. + /// Prepares a child node with a specified timestamp. /// /// # Arguments /// /// * `payload` - The payload data for the node /// * `parents` - The parent CIDs for this node /// * `genesis` - The genesis CID this node belongs to - /// * `timestamp` - The timestamp to use (from the original node) + /// * `timestamp` - The timestamp for CID generation /// * `metadata` - The metadata for the node /// /// # Returns /// /// A tuple of (CID, Node) for the created child node - pub fn prepare_child_node_with_timestamp( + pub fn prepare_child_node( &mut self, payload: P, parents: Vec, @@ -239,13 +249,6 @@ 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_nanos() as u64) - } /// Check if adding an edge (new node with parents) would create a cycle fn would_create_cycle_with(&mut self, new_cid: &Cid, parents: &[Cid]) -> Result { @@ -812,7 +815,7 @@ mod tests { #[test] fn test_add_genesis_node() { let mut dag = DagGraph::new(MockStorage::new()); - let cid = dag.add_genesis_node("test".to_string(), ()).unwrap(); + let cid = dag.add_genesis_node("test".to_string(), 1000, ()).unwrap(); let latest = dag.calculate_latest(&cid).unwrap(); assert_eq!(latest, Some(cid)); @@ -821,9 +824,11 @@ mod tests { #[test] fn test_add_child_node() { let mut dag = DagGraph::new(MockStorage::new()); - let genesis_cid = dag.add_genesis_node("genesis".to_string(), ()).unwrap(); + let genesis_cid = dag + .add_genesis_node("genesis".to_string(), 1000, ()) + .unwrap(); let child_cid = dag - .add_child_node("child".to_string(), vec![genesis_cid], genesis_cid, ()) + .add_child_node("child".to_string(), vec![genesis_cid], genesis_cid, 2000, ()) .unwrap(); let latest = dag.calculate_latest(&genesis_cid).unwrap(); @@ -878,7 +883,7 @@ mod tests { // Add a new node whose parent is B (should NOT create a cycle) let new_cid = dag - .add_node("payload".to_string(), vec![cid_b], BTreeMap::new()) + .add_node("payload".to_string(), vec![cid_b], 3000, BTreeMap::new()) .expect("add_node should succeed"); // Verify edges_forward is updated (B -> new_cid) @@ -901,13 +906,13 @@ mod tests { // The first add_node call builds the cache let cid1 = dag - .add_node("n1".to_string(), vec![cid_b], BTreeMap::new()) + .add_node("n1".to_string(), vec![cid_b], 3000, BTreeMap::new()) .expect("first add"); let cache_size_before = dag.edges_forward.len(); // The second add_node call reuses the cache let _cid2 = dag - .add_node("n2".to_string(), vec![cid1], BTreeMap::new()) + .add_node("n2".to_string(), vec![cid1], 4000, BTreeMap::new()) .expect("second add"); // One extra node -> cache size should increase by exactly 1 @@ -1064,7 +1069,7 @@ mod tests { let mut dag = DagGraph::new(storage); let genesis = dag - .add_genesis_node("payload".to_string(), BTreeMap::new()) + .add_genesis_node("payload".to_string(), 1000, BTreeMap::new()) .unwrap(); dag.remove_node(&genesis).unwrap(); @@ -1078,9 +1083,9 @@ mod tests { let mut dag = DagGraph::new(storage); let genesis = dag - .add_genesis_node("payload".to_string(), BTreeMap::new()) + .add_genesis_node("payload".to_string(), 1000, BTreeMap::new()) .unwrap(); - dag.add_child_node("child".to_string(), vec![genesis], genesis, BTreeMap::new()) + dag.add_child_node("child".to_string(), vec![genesis], genesis, 2000, BTreeMap::new()) .unwrap(); let err = dag.remove_node(&genesis); @@ -1094,7 +1099,7 @@ mod tests { let mut dag = DagGraph::new(storage); let (genesis_cid, genesis_node) = dag - .prepare_genesis_node("payload".to_string(), BTreeMap::new()) + .prepare_genesis_node("payload".to_string(), 1000, BTreeMap::new()) .unwrap(); dag.storage.put(&genesis_node).unwrap(); dag.register_prepared_node(genesis_cid, &genesis_node) @@ -1106,6 +1111,7 @@ mod tests { "child".to_string(), vec![genesis_cid], genesis_cid, + 2000, BTreeMap::new(), ) .unwrap(); diff --git a/src/repo.rs b/src/repo.rs index ce4a1bf..3e87cc3 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::crdt::timestamp::next_monotonic_timestamp; use crate::storage::{BatchError, LeveldbBatchGuard, SharedLeveldb, SharedLeveldbAccess}; use crate::{ crdt::{ @@ -54,77 +55,35 @@ where } } - pub fn commit_operation(&mut self, op: Operation) -> Result { - // Recently, Merge operations cannot be manually committed - if matches!(op.kind, OperationType::Merge(_)) { - return Err(CrdtError::Internal( - "Merge operations cannot be manually committed".to_string(), - )); - } - - self.commit_operation_internal(op, false) - } - - /// Imports an operation from another replica with the original node timestamp. + /// Commits an operation to the repository. /// - /// Unlike `commit_operation`, this method preserves the original timestamp - /// from the source replica, ensuring that the resulting CID matches the - /// original. This is essential for CRDT replication where CID consistency - /// across replicas is required. + /// If `op.node_timestamp` is set, the operation is treated as an import from + /// another replica, preserving the original timestamp for CID consistency. + /// Otherwise, the current time is used for the DAG node timestamp. /// /// # Arguments /// - /// * `op` - The operation to import (with genesis already set for Create operations) - /// * `node_timestamp` - The original DAG node timestamp from the source replica + /// * `op` - The operation to commit /// /// # Returns /// - /// The CID of the imported node (should match the original CID) + /// The CID of the committed node /// /// # Errors /// - /// Returns an error if the operation cannot be applied or if there are - /// consistency issues with the DAG structure. - pub fn import_operation( - &mut self, - op: Operation, - node_timestamp: u64, - ) -> Result { - let shared = self.shared_leveldb()?; - let batch_guard = Self::begin_shared_batch(&shared)?; - let mut pending_nodes: Vec = Vec::new(); - - let cid = match op.kind.clone() { - OperationType::Create(payload) => { - self.stage_create_with_timestamp(payload, &op, node_timestamp, &mut pending_nodes)? - } - OperationType::Update(payload) => { - self.stage_update_with_timestamp( - payload, - &op, - node_timestamp, - &mut pending_nodes, - )? - } - OperationType::Delete => { - self.stage_delete_with_timestamp(&op, node_timestamp, &mut pending_nodes)? - } - OperationType::Merge(payload) => { - self.stage_merge_with_timestamp(payload, &op, node_timestamp, &mut pending_nodes)? - } - }; - - 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)); + /// Returns an error if: + /// - Merge operations are attempted to be committed manually (without node_timestamp) + /// - The operation cannot be applied + /// - There are consistency issues with the DAG structure + pub fn commit_operation(&mut self, op: Operation) -> Result { + // Merge operations can only be committed via import (with node_timestamp) or auto-merge + if matches!(op.kind, OperationType::Merge(_)) && op.node_timestamp.is_none() { + return Err(CrdtError::Internal( + "Merge operations cannot be manually committed".to_string(), + )); } - Ok(cid) + self.commit_operation_internal(op, false) } pub fn latest(&self, genesis_id: &Cid) -> Option { @@ -242,22 +201,58 @@ where let batch_guard = Self::begin_shared_batch(&shared)?; let mut pending_nodes: Vec = Vec::new(); - if !skip_auto_merge { + // If node_timestamp is set, this is an import operation - skip auto-merge + let is_import = op.node_timestamp.is_some(); + if !skip_auto_merge && !is_import { self.ensure_parent_context(&mut op, &mut pending_nodes)?; } - let cid = match op.kind.clone() { - OperationType::Create(payload) => { - self.stage_create(payload, &mut op, &mut pending_nodes)? - } - OperationType::Update(payload) => { - self.stage_update(payload, &op, &mut pending_nodes)? + let cid = if let Some(node_timestamp) = op.node_timestamp { + // Import path: use the specified timestamp for CID consistency + match op.kind.clone() { + OperationType::Create(payload) => { + self.stage_create_with_timestamp( + payload, + &op, + node_timestamp, + &mut pending_nodes, + )? + } + OperationType::Update(payload) => { + self.stage_update_with_timestamp( + payload, + &op, + node_timestamp, + &mut pending_nodes, + )? + } + OperationType::Delete => { + self.stage_delete_with_timestamp(&op, node_timestamp, &mut pending_nodes)? + } + OperationType::Merge(payload) => { + self.stage_merge_with_timestamp( + payload, + &op, + node_timestamp, + &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(), - )) + } else { + // Normal path: use current time for timestamp + match op.kind.clone() { + OperationType::Create(payload) => { + self.stage_create(payload, &mut op, &mut pending_nodes)? + } + 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(), + )) + } } }; @@ -335,9 +330,10 @@ where op: &mut Operation, pending_nodes: &mut Vec, ) -> Result { + let timestamp = next_monotonic_timestamp(); let (genesis_cid, node) = self .dag - .prepare_genesis_node(payload, ContentMetadata::default())?; + .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; let cid = self.stage_prepared_node(genesis_cid, node, pending_nodes)?; op.genesis = cid; Ok(cid) @@ -354,7 +350,7 @@ where ) -> Result { let (genesis_cid, node) = self .dag - .prepare_genesis_node_with_timestamp(payload, timestamp, ContentMetadata::default())?; + .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; // Verify that the computed CID matches the expected genesis from the operation if genesis_cid != op.genesis { @@ -373,11 +369,16 @@ where op: &Operation, pending_nodes: &mut Vec, ) -> Result { + let timestamp = next_monotonic_timestamp(); 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)?; + let (cid, node) = self.dag.prepare_child_node( + payload, + op.parents.clone(), + op.genesis, + timestamp, + metadata, + )?; self.stage_prepared_node(cid, node, pending_nodes) } @@ -391,7 +392,7 @@ where ) -> Result { let metadata = self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; - let (cid, node) = self.dag.prepare_child_node_with_timestamp( + let (cid, node) = self.dag.prepare_child_node( payload, op.parents.clone(), op.genesis, @@ -406,6 +407,7 @@ where op: &Operation, pending_nodes: &mut Vec, ) -> Result { + let timestamp = next_monotonic_timestamp(); let ops = self.state.get_operations_by_genesis(&op.genesis)?; let last_payload = ops .iter() @@ -415,7 +417,7 @@ where .cloned() .map(|payload| (operation.timestamp, payload)) }) - .max_by_key(|(timestamp, _)| *timestamp) + .max_by_key(|(ts, _)| *ts) .map(|(_, payload)| payload) .ok_or_else(|| { CrdtError::Internal(format!( @@ -426,9 +428,13 @@ where 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)?; + let (cid, node) = self.dag.prepare_child_node( + last_payload, + op.parents.clone(), + op.genesis, + timestamp, + metadata, + )?; self.stage_prepared_node(cid, node, pending_nodes) } @@ -459,7 +465,7 @@ where let metadata = self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; - let (cid, node) = self.dag.prepare_child_node_with_timestamp( + let (cid, node) = self.dag.prepare_child_node( last_payload, op.parents.clone(), op.genesis, @@ -479,7 +485,7 @@ where ) -> Result { let metadata = self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; - let (cid, node) = self.dag.prepare_child_node_with_timestamp( + let (cid, node) = self.dag.prepare_child_node( payload, op.parents.clone(), op.genesis, @@ -548,18 +554,24 @@ where let policy_type = genesis_node.metadata().policy_type(); let policy = self.create_policy(policy_type)?; - let merge_node = - self.resolver - .create_merge_node(&heads, &self.dag, *genesis, policy.as_ref())?; - self.validate_parent_genesis(genesis, &heads)?; + let merge_timestamp = next_monotonic_timestamp(); + let merge_node = self.resolver.create_merge_node( + &heads, + &self.dag, + *genesis, + merge_timestamp, + policy.as_ref(), + )?; + let (merge_cid, node) = self .dag .prepare_child_node( merge_node.payload().clone(), heads.clone(), *genesis, + merge_timestamp, merge_node.metadata().clone(), ) .map_err(CrdtError::Graph)?; @@ -1257,6 +1269,7 @@ mod tests { .dag .prepare_genesis_node( TestPayload("dangling".to_string()), + 1000, ContentMetadata::default(), ) .unwrap(); @@ -1561,6 +1574,7 @@ mod tests { branch_a_payload.clone(), vec![genesis], genesis, + 2000, ContentMetadata::default(), ) .unwrap(); @@ -1579,6 +1593,7 @@ mod tests { branch_b_payload.clone(), vec![genesis], genesis, + 3000, ContentMetadata::default(), ) .unwrap(); @@ -1619,6 +1634,7 @@ mod tests { branch_a_payload.clone(), vec![genesis], genesis, + 2000, ContentMetadata::default(), ) .unwrap(); @@ -1639,6 +1655,7 @@ mod tests { branch_b_payload.clone(), vec![genesis], genesis, + 3000, ContentMetadata::default(), ) .unwrap(); @@ -1657,6 +1674,7 @@ mod tests { merge_payload.clone(), vec![branch_a, branch_b], genesis, + 4000, ContentMetadata::default(), ) .unwrap(); @@ -1677,6 +1695,7 @@ mod tests { latest_payload.clone(), vec![merge_cid], genesis, + 5000, ContentMetadata::default(), ) .unwrap(); @@ -1722,12 +1741,13 @@ mod tests { let node = repo1.dag.get_node(&cid1).unwrap().unwrap(); let node_timestamp = node.timestamp(); - // Create the operation with the correct genesis CID for import + // Create the operation with the correct genesis CID and node_timestamp for import let mut import_op = make_test_operation(cid1, OperationType::Create(payload)); import_op.genesis = cid1; + import_op.node_timestamp = Some(node_timestamp); - // Import the operation into repo2 with the original timestamp - let cid2 = repo2.import_operation(import_op, node_timestamp).unwrap(); + // Import the operation into repo2 + let cid2 = repo2.commit_operation(import_op).unwrap(); // CIDs should match assert_eq!(cid1, cid2, "CIDs should be identical after import"); @@ -1763,9 +1783,8 @@ mod tests { let mut import_create_op = make_test_operation(genesis_cid, OperationType::Create(create_payload)); import_create_op.genesis = genesis_cid; - let imported_genesis = repo2 - .import_operation(import_create_op, genesis_timestamp) - .unwrap(); + import_create_op.node_timestamp = Some(genesis_timestamp); + let imported_genesis = repo2.commit_operation(import_create_op).unwrap(); assert_eq!(genesis_cid, imported_genesis); // Create update in repo1 @@ -1784,9 +1803,8 @@ mod tests { let mut import_update_op = make_test_operation(genesis_cid, OperationType::Update(update_payload)); import_update_op.parents = update_parents; - let imported_update = repo2 - .import_operation(import_update_op, update_timestamp) - .unwrap(); + import_update_op.node_timestamp = Some(update_timestamp); + let imported_update = repo2.commit_operation(import_update_op).unwrap(); // CIDs should match assert_eq!( @@ -1810,9 +1828,10 @@ mod tests { let payload = TestPayload("test content".to_string()); let mut op = make_test_operation(wrong_genesis, OperationType::Create(payload)); op.genesis = wrong_genesis; // This won't match the computed CID + op.node_timestamp = Some(12345); // Set node_timestamp to trigger import path // Import should fail due to CID mismatch - let result = repo.import_operation(op, 12345); + let result = repo.commit_operation(op); assert!(result.is_err()); match result { Err(CrdtError::Internal(msg)) => { From 16024ceaa9fc73b61d20ed536a83099234185ad8 Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 15:08:51 +0900 Subject: [PATCH 3/7] Stage create/update/delete/merge operations with timestamp --- src/repo.rs | 239 ++++++++++++++++------------------------------------ 1 file changed, 72 insertions(+), 167 deletions(-) diff --git a/src/repo.rs b/src/repo.rs index 3e87cc3..997bec3 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -207,52 +207,26 @@ where self.ensure_parent_context(&mut op, &mut pending_nodes)?; } - let cid = if let Some(node_timestamp) = op.node_timestamp { - // Import path: use the specified timestamp for CID consistency - match op.kind.clone() { - OperationType::Create(payload) => { - self.stage_create_with_timestamp( - payload, - &op, - node_timestamp, - &mut pending_nodes, - )? - } - OperationType::Update(payload) => { - self.stage_update_with_timestamp( - payload, - &op, - node_timestamp, - &mut pending_nodes, - )? - } - OperationType::Delete => { - self.stage_delete_with_timestamp(&op, node_timestamp, &mut pending_nodes)? - } - OperationType::Merge(payload) => { - self.stage_merge_with_timestamp( - payload, - &op, - node_timestamp, - &mut pending_nodes, - )? - } + // Use specified timestamp or generate a new one + let timestamp = op.node_timestamp.unwrap_or_else(next_monotonic_timestamp); + + let cid = match op.kind.clone() { + OperationType::Create(payload) => { + self.stage_create(payload, &mut op, timestamp, is_import, &mut pending_nodes)? } - } else { - // Normal path: use current time for timestamp - match op.kind.clone() { - OperationType::Create(payload) => { - self.stage_create(payload, &mut op, &mut pending_nodes)? - } - OperationType::Update(payload) => { - self.stage_update(payload, &op, &mut pending_nodes)? - } - OperationType::Delete => self.stage_delete(&op, &mut pending_nodes)?, - OperationType::Merge(_) => { + OperationType::Update(payload) => { + self.stage_update(payload, &op, timestamp, is_import, &mut pending_nodes)? + } + OperationType::Delete => { + self.stage_delete(&op, timestamp, is_import, &mut pending_nodes)? + } + OperationType::Merge(payload) => { + if !is_import { return Err(CrdtError::Internal( "Merge operations must be committed via auto-merge".to_string(), - )) + )); } + self.stage_merge(payload, &op, timestamp, &mut pending_nodes)? } }; @@ -324,74 +298,52 @@ where Ok(()) } + /// Stages a Create operation. + /// + /// # Arguments + /// * `payload` - The payload for the new content + /// * `op` - The operation (genesis will be set if not importing) + /// * `timestamp` - The timestamp to use for CID generation + /// * `is_import` - If true, verifies CID matches op.genesis; if false, sets op.genesis + /// * `pending_nodes` - Accumulator for pending nodes fn stage_create( &mut self, payload: Payload, op: &mut Operation, - pending_nodes: &mut Vec, - ) -> Result { - let timestamp = next_monotonic_timestamp(); - let (genesis_cid, node) = self - .dag - .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; - let cid = self.stage_prepared_node(genesis_cid, node, pending_nodes)?; - op.genesis = cid; - Ok(cid) - } - - /// Stages a Create operation with a specified timestamp (for replication). - /// Unlike `stage_create`, this preserves the original genesis CID from the operation. - fn stage_create_with_timestamp( - &mut self, - payload: Payload, - op: &Operation, timestamp: u64, + is_import: bool, pending_nodes: &mut Vec, ) -> Result { let (genesis_cid, node) = self .dag .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; - // Verify that the computed CID matches the expected genesis from the operation - if genesis_cid != op.genesis { - return Err(CrdtError::Internal(format!( - "CID mismatch during import: expected {}, got {}", - op.genesis, genesis_cid - ))); + if is_import { + // Verify that the computed CID matches the expected genesis from the operation + if genesis_cid != op.genesis { + return Err(CrdtError::Internal(format!( + "CID mismatch during import: expected {}, got {}", + op.genesis, genesis_cid + ))); + } + } else { + op.genesis = genesis_cid; } self.stage_prepared_node(genesis_cid, node, pending_nodes) } + /// Stages an Update operation. fn stage_update( - &mut self, - payload: Payload, - op: &Operation, - pending_nodes: &mut Vec, - ) -> Result { - let timestamp = next_monotonic_timestamp(); - 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, - timestamp, - metadata, - )?; - self.stage_prepared_node(cid, node, pending_nodes) - } - - /// Stages an Update operation with a specified timestamp (for replication). - fn stage_update_with_timestamp( &mut self, payload: Payload, op: &Operation, timestamp: u64, + is_import: bool, pending_nodes: &mut Vec, ) -> Result { let metadata = - self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), is_import)?; let (cid, node) = self.dag.prepare_child_node( payload, op.parents.clone(), @@ -402,47 +354,12 @@ where self.stage_prepared_node(cid, node, pending_nodes) } + /// Stages a Delete operation. fn stage_delete( - &mut self, - op: &Operation, - pending_nodes: &mut Vec, - ) -> Result { - let timestamp = next_monotonic_timestamp(); - 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(|(ts, _)| *ts) - .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, - timestamp, - metadata, - )?; - self.stage_prepared_node(cid, node, pending_nodes) - } - - /// Stages a Delete operation with a specified timestamp (for replication). - fn stage_delete_with_timestamp( &mut self, op: &Operation, timestamp: u64, + is_import: bool, pending_nodes: &mut Vec, ) -> Result { let ops = self.state.get_operations_by_genesis(&op.genesis)?; @@ -464,7 +381,7 @@ where })?; let metadata = - self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), is_import)?; let (cid, node) = self.dag.prepare_child_node( last_payload, op.parents.clone(), @@ -475,16 +392,17 @@ where self.stage_prepared_node(cid, node, pending_nodes) } - /// Stages a Merge operation with a specified timestamp (for replication). - fn stage_merge_with_timestamp( + /// Stages a Merge operation (only for imports). + fn stage_merge( &mut self, payload: Payload, op: &Operation, timestamp: u64, pending_nodes: &mut Vec, ) -> Result { + // Merge operations are always imports, so use lenient metadata resolution let metadata = - self.resolve_metadata_for_import(&op.genesis, &op.parents, pending_nodes.as_slice())?; + self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), true)?; let (cid, node) = self.dag.prepare_child_node( payload, op.parents.clone(), @@ -629,51 +547,34 @@ where } } - fn resolve_metadata_for_commit( - &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) - .map_err(CrdtError::Graph)? - .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) - .map_err(CrdtError::Graph)? - .ok_or_else(|| CrdtError::Internal(format!("Genesis not found: {genesis}")))?; - Ok(genesis_node.metadata().clone()) - } - } - - /// Resolves metadata for import operations. - /// This is more lenient than `resolve_metadata_for_commit` as parent nodes - /// may not exist yet during replication (operations may arrive out of order). - fn resolve_metadata_for_import( + /// Resolves metadata for an operation. + /// + /// # Arguments + /// * `genesis` - The genesis CID + /// * `parents` - The parent CIDs + /// * `pending_nodes` - Pending nodes that haven't been committed yet + /// * `lenient` - If true, returns default metadata when nodes not found (for imports) + fn resolve_metadata( &self, genesis: &Cid, parents: &[Cid], pending_nodes: &[PendingNode], + lenient: bool, ) -> Result { // Try to get metadata from parents first if let Some(parent) = parents.first() { if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == parent) { return Ok(pending.metadata.clone()); } - if let Ok(Some(node)) = self.dag.get_node(parent) { - return Ok(node.metadata().clone()); + match self.dag.get_node(parent) { + Ok(Some(node)) => return Ok(node.metadata().clone()), + Ok(None) if !lenient => { + return Err(CrdtError::Internal(format!( + "Parent node not found: {parent}" + ))) + } + Err(e) if !lenient => return Err(CrdtError::Graph(e)), + _ => {} // lenient mode: continue to try genesis } } @@ -681,12 +582,16 @@ where if let Some(pending) = pending_nodes.iter().find(|pending| &pending.cid == genesis) { return Ok(pending.metadata.clone()); } - if let Ok(Some(genesis_node)) = self.dag.get_node(genesis) { - return Ok(genesis_node.metadata().clone()); + match self.dag.get_node(genesis) { + Ok(Some(genesis_node)) => Ok(genesis_node.metadata().clone()), + Ok(None) if lenient => Ok(ContentMetadata::default()), + Ok(None) => Err(CrdtError::Internal(format!("Genesis not found: {genesis}"))), + Err(_) if lenient => { + // In lenient mode, return default metadata on error + Ok(ContentMetadata::default()) + } + Err(e) => Err(CrdtError::Graph(e)), } - - // Default metadata if nothing is found (for out-of-order imports) - Ok(ContentMetadata::default()) } fn node_characteristics(&self, cid: &Cid) -> Result<(bool, u64)> { let node = self From 322cc5b5f8c333e7f39fbfd571de9319ac0b2bf7 Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 15:12:33 +0900 Subject: [PATCH 4/7] Refactor create, update, delete operations handling --- src/repo.rs | 35 +++++++++++++++-------------------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/src/repo.rs b/src/repo.rs index 997bec3..9ea6450 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -201,9 +201,8 @@ where let batch_guard = Self::begin_shared_batch(&shared)?; let mut pending_nodes: Vec = Vec::new(); - // If node_timestamp is set, this is an import operation - skip auto-merge - let is_import = op.node_timestamp.is_some(); - if !skip_auto_merge && !is_import { + // If node_timestamp is not set, run auto-merge logic + if !skip_auto_merge && op.node_timestamp.is_none() { self.ensure_parent_context(&mut op, &mut pending_nodes)?; } @@ -212,16 +211,16 @@ where let cid = match op.kind.clone() { OperationType::Create(payload) => { - self.stage_create(payload, &mut op, timestamp, is_import, &mut pending_nodes)? + self.stage_create(payload, &mut op, timestamp, &mut pending_nodes)? } OperationType::Update(payload) => { - self.stage_update(payload, &op, timestamp, is_import, &mut pending_nodes)? + self.stage_update(payload, &op, timestamp, &mut pending_nodes)? } OperationType::Delete => { - self.stage_delete(&op, timestamp, is_import, &mut pending_nodes)? + self.stage_delete(&op, timestamp, &mut pending_nodes)? } OperationType::Merge(payload) => { - if !is_import { + if op.node_timestamp.is_none() { return Err(CrdtError::Internal( "Merge operations must be committed via auto-merge".to_string(), )); @@ -300,26 +299,21 @@ where /// Stages a Create operation. /// - /// # Arguments - /// * `payload` - The payload for the new content - /// * `op` - The operation (genesis will be set if not importing) - /// * `timestamp` - The timestamp to use for CID generation - /// * `is_import` - If true, verifies CID matches op.genesis; if false, sets op.genesis - /// * `pending_nodes` - Accumulator for pending nodes + /// If `op.node_timestamp` is set (import), verifies CID matches op.genesis. + /// Otherwise, sets op.genesis to the computed CID. fn stage_create( &mut self, payload: Payload, op: &mut Operation, timestamp: u64, - is_import: bool, pending_nodes: &mut Vec, ) -> Result { let (genesis_cid, node) = self .dag .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; - if is_import { - // Verify that the computed CID matches the expected genesis from the operation + if op.node_timestamp.is_some() { + // Import: verify that the computed CID matches the expected genesis if genesis_cid != op.genesis { return Err(CrdtError::Internal(format!( "CID mismatch during import: expected {}, got {}", @@ -327,6 +321,7 @@ where ))); } } else { + // Local create: set genesis to the computed CID op.genesis = genesis_cid; } @@ -339,11 +334,11 @@ where payload: Payload, op: &Operation, timestamp: u64, - is_import: bool, pending_nodes: &mut Vec, ) -> Result { + let lenient = op.node_timestamp.is_some(); let metadata = - self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), is_import)?; + self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), lenient)?; let (cid, node) = self.dag.prepare_child_node( payload, op.parents.clone(), @@ -359,7 +354,6 @@ where &mut self, op: &Operation, timestamp: u64, - is_import: bool, pending_nodes: &mut Vec, ) -> Result { let ops = self.state.get_operations_by_genesis(&op.genesis)?; @@ -380,8 +374,9 @@ where )) })?; + let lenient = op.node_timestamp.is_some(); let metadata = - self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), is_import)?; + self.resolve_metadata(&op.genesis, &op.parents, pending_nodes.as_slice(), lenient)?; let (cid, node) = self.dag.prepare_child_node( last_payload, op.parents.clone(), From dbb90a4fef2031c2c6fcaa99097c8959d09c1098 Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 15:16:45 +0900 Subject: [PATCH 5/7] Update version to 0.1.0 in Cargo.toml --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 96eb801..4ac928d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -250,7 +250,7 @@ checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" [[package]] name = "crsl-lib" -version = "0.2.0" +version = "0.1.0" dependencies = [ "bincode", "cid", diff --git a/Cargo.toml b/Cargo.toml index f1dbdc8..43cddc8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "crsl-lib" -version = "0.2.0" +version = "0.1.0" edition = "2021" rust-version = "1.79" From abb27d1ee9ca4e2cfc8ecf8f233feb767dd377fc Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 15:17:08 +0900 Subject: [PATCH 6/7] Refactor prepare_child_node calls for readability --- src/graph/dag.rs | 25 +++++++++++++++++++------ src/repo.rs | 16 ++++++++-------- 2 files changed, 27 insertions(+), 14 deletions(-) diff --git a/src/graph/dag.rs b/src/graph/dag.rs index c30bffb..05ef90f 100644 --- a/src/graph/dag.rs +++ b/src/graph/dag.rs @@ -81,7 +81,8 @@ where GraphError::Internal("child node requires at least one parent".to_string()) })?; - let (cid, node) = self.prepare_child_node(payload, parents, genesis, timestamp, metadata)?; + let (cid, node) = + self.prepare_child_node(payload, parents, genesis, timestamp, metadata)?; self.persist_and_cache(cid, node) } @@ -114,7 +115,8 @@ where timestamp: u64, metadata: M, ) -> Result { - let (cid, node) = self.prepare_child_node(payload, parents, genesis, timestamp, metadata)?; + let (cid, node) = + self.prepare_child_node(payload, parents, genesis, timestamp, metadata)?; self.persist_and_cache(cid, node) } @@ -249,7 +251,6 @@ where Ok(result) } - /// Check if adding an edge (new node with parents) would create a cycle fn would_create_cycle_with(&mut self, new_cid: &Cid, parents: &[Cid]) -> Result { // Build cache only for the relevant subgraph @@ -828,7 +829,13 @@ mod tests { .add_genesis_node("genesis".to_string(), 1000, ()) .unwrap(); let child_cid = dag - .add_child_node("child".to_string(), vec![genesis_cid], genesis_cid, 2000, ()) + .add_child_node( + "child".to_string(), + vec![genesis_cid], + genesis_cid, + 2000, + (), + ) .unwrap(); let latest = dag.calculate_latest(&genesis_cid).unwrap(); @@ -1085,8 +1092,14 @@ mod tests { let genesis = dag .add_genesis_node("payload".to_string(), 1000, BTreeMap::new()) .unwrap(); - dag.add_child_node("child".to_string(), vec![genesis], genesis, 2000, BTreeMap::new()) - .unwrap(); + dag.add_child_node( + "child".to_string(), + vec![genesis], + genesis, + 2000, + BTreeMap::new(), + ) + .unwrap(); let err = dag.remove_node(&genesis); assert!(err.is_err()); diff --git a/src/repo.rs b/src/repo.rs index 9ea6450..57042a2 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -216,9 +216,7 @@ where OperationType::Update(payload) => { self.stage_update(payload, &op, timestamp, &mut pending_nodes)? } - OperationType::Delete => { - self.stage_delete(&op, timestamp, &mut pending_nodes)? - } + OperationType::Delete => self.stage_delete(&op, timestamp, &mut pending_nodes)?, OperationType::Merge(payload) => { if op.node_timestamp.is_none() { return Err(CrdtError::Internal( @@ -308,9 +306,9 @@ where timestamp: u64, pending_nodes: &mut Vec, ) -> Result { - let (genesis_cid, node) = self - .dag - .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; + let (genesis_cid, node) = + self.dag + .prepare_genesis_node(payload, timestamp, ContentMetadata::default())?; if op.node_timestamp.is_some() { // Import: verify that the computed CID matches the expected genesis @@ -1671,8 +1669,10 @@ mod tests { multihash::Multihash::<64>::wrap(0x12, b"import-update-test").unwrap(), ); let create_payload = TestPayload("initial".to_string()); - let create_op = - make_test_operation(initial_genesis, OperationType::Create(create_payload.clone())); + let create_op = make_test_operation( + initial_genesis, + OperationType::Create(create_payload.clone()), + ); let genesis_cid = repo1.commit_operation(create_op.clone()).unwrap(); // Get genesis node timestamp From eb654c8e9a35fa1266fbcbdef6736ee78b328a57 Mon Sep 17 00:00:00 2001 From: somasekimoto <0421.soma@gmail.com> Date: Sun, 14 Dec 2025 15:37:38 +0900 Subject: [PATCH 7/7] Remove constructor for operation timestamp preservation --- src/crdt/operation.rs | 34 ---------------------------------- 1 file changed, 34 deletions(-) diff --git a/src/crdt/operation.rs b/src/crdt/operation.rs index 756c040..846bc66 100644 --- a/src/crdt/operation.rs +++ b/src/crdt/operation.rs @@ -98,40 +98,6 @@ where } } - /// Creates a new operation with a specified node timestamp (for replication). - /// - /// When importing operations from other replicas, use this constructor - /// to preserve the original node timestamp for CID consistency. - /// - /// # Arguments - /// - /// * `genesis` - ID of the content being operated on - /// * `kind` - Type of operation and its payload - /// * `author` - User/system performing the operation - /// * `node_timestamp` - The original DAG node timestamp from the source replica - /// - /// # Returns - /// - /// A newly created operation object with the specified node timestamp - pub fn with_node_timestamp( - genesis: ContentId, - kind: OperationType, - author: Author, - node_timestamp: u64, - ) -> Self { - let timestamp = next_monotonic_timestamp(); - let id = Ulid::new(); - Self { - id, - genesis, - kind, - timestamp, - author, - parents: Vec::new(), - node_timestamp: Some(node_timestamp), - } - } - /// Checks if this operation is of the given kind pub fn is_type(&self, kind: OperationKind) -> bool { self.kind.as_kind() == kind