feat: Add timestamp argument for deterministic CID generation

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<S>(
         &self,
         heads: &[Cid],
         dag: &DagGraph<S, P, M>,
         genesis: Cid,
+        timestamp: u64,
         policy: &dyn MergePolicy<P>,
     ) -> CrdtResult<Node<P, M>>
     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<u64> {
-        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::<String, ContentMetadata>::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,

src/crdt/operation.rs

@@ -61,6 +61,11 @@ pub struct Operation<ContentId, T> {
     pub author: Author,
     #[serde(default = "Vec::new")]
     pub parents: Vec<ContentId>,
+    /// 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<Timestamp>,
 }
 
 impl<ContentId, T> Operation<ContentId, T>
@@ -89,6 +94,7 @@ where
             timestamp,
             author,
             parents: Vec::new(),
+            node_timestamp: None,
         }
     }
 

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,
         }
     }
 

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<Cid>, metadata: M) -> Result<Cid> {
+    /// 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<Cid>,
+        timestamp: u64,
+        metadata: M,
+    ) -> Result<Cid> {
         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,41 +81,88 @@ 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<Cid> {
-        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<Cid> {
+        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<Cid>,
         genesis: Cid,
+        timestamp: u64,
         metadata: M,
     ) -> Result<Cid> {
-        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<P, M>)> {
-        let timestamp = Self::current_timestamp()?;
+    /// Prepares a genesis node with a specified timestamp.
+    ///
+    /// # Arguments
+    ///
+    /// * `payload` - The payload data for the 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(
+        &mut self,
+        payload: P,
+        timestamp: u64,
+        metadata: M,
+    ) -> Result<(Cid, Node<P, M>)> {
         let node = Node::new_genesis(payload, timestamp, metadata);
         let cid = node.content_id()?;
         Ok((cid, node))
     }
 
+    /// 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 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(
         &mut self,
         payload: P,
         parents: Vec<Cid>,
         genesis: Cid,
+        timestamp: u64,
         metadata: M,
     ) -> Result<(Cid, Node<P, M>)> {
-        let timestamp = Self::current_timestamp()?;
         let node = Node::new_child(payload, parents.clone(), genesis, timestamp, metadata);
         let cid = node.content_id()?;
 
@@ -189,14 +251,6 @@ where
         Ok(result)
     }
 
-    /// Returns the current time in nanoseconds since the Unix epoch.
-    fn current_timestamp() -> Result<u64> {
-        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<bool> {
         // Build cache only for the relevant subgraph
@@ -762,7 +816,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));
@@ -771,9 +825,17 @@ 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();
@@ -828,7 +890,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)
@@ -851,13 +913,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
@@ -1014,7 +1076,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();
@@ -1028,10 +1090,16 @@ mod tests {
         let mut dag = DagGraph::new(storage);
 
         let genesis = dag
-            .add_genesis_node("payload".to_string(), BTreeMap::new())
-            .unwrap();
-        dag.add_child_node("child".to_string(), vec![genesis], genesis, BTreeMap::new())
+            .add_genesis_node("payload".to_string(), 1000, 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());
@@ -1044,7 +1112,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)
@@ -1056,6 +1124,7 @@ mod tests {
                 "child".to_string(),
                 vec![genesis_cid],
                 genesis_cid,
+                2000,
                 BTreeMap::new(),
             )
             .unwrap();