fix dag initialized

consensus/src/dag/blockdag.rs

@@ -19,6 +19,7 @@ use starcoin_accumulator::node::AccumulatorStoreType;
 use starcoin_accumulator::{Accumulator, MerkleAccumulator};
 use starcoin_config::{NodeConfig, RocksdbConfig};
 use starcoin_crypto::{HashValue as Hash, HashValue};
+use starcoin_logger::prelude::info;
 use starcoin_storage::flexi_dag::SyncFlexiDagSnapshotHasher;
 use starcoin_storage::storage::CodecKVStore;
 use starcoin_storage::Store;

flexidag/src/flexidag_service.rs

@@ -295,6 +295,9 @@ impl ServiceFactory<Self> for FlexidagService {
         let config = ctx.get_shared::<Arc<NodeConfig>>()?;
         let (dag, dag_accumulator) =
             BlockDAG::try_init_with_storage(storage.clone(), config.clone())?;
+        if let Some(dag) = &dag {
+            ctx.put_shared(dag.clone())?;
+        }
         let tip_info = dag_accumulator.as_ref().map(|accumulator| {
             let tips_index = accumulator.num_leaves();
             let tips_key = accumulator

node/Cargo.toml

@@ -49,6 +49,7 @@ timeout-join-handler = { workspace = true }
 tokio = { features = ["full"], workspace = true }
 starcoin-accumulator = { workspace = true }
 num_cpus = { workspace = true }
+starcoin-flexidag = { workspace = true }
 
 [dev-dependencies]
 stest = { workspace = true }

node/src/node.rs

@@ -20,6 +20,7 @@ use starcoin_chain_service::ChainReaderService;
 use starcoin_config::NodeConfig;
 use starcoin_consensus::{BlockDAG, FlexiDagStorage, FlexiDagStorageConfig};
 use starcoin_crypto::HashValue;
+use starcoin_flexidag::FlexidagService;
 use starcoin_genesis::{Genesis, GenesisError};
 use starcoin_logger::prelude::*;
 use starcoin_logger::structured_log::init_slog_logger;
@@ -357,16 +358,12 @@ impl NodeService {
         let upgrade_time = SystemTime::now().duration_since(start_time)?;
         let storage = Arc::new(Storage::new(storage_instance)?);
         registry.put_shared(storage.clone()).await?;
-        let dag_storage = FlexiDagStorage::create_from_path(
-            config.storage.dag_dir(),
-            config.storage.clone().into(),
-        )?;
-        let dag = BlockDAG::new(8, dag_storage);
+
         let (chain_info, genesis) =
             Genesis::init_and_check_storage(config.net(), storage.clone(), config.data_dir())?;
-        // TODO: init dag in the dag fork height
-        let _ = dag.init_with_genesis(genesis.block().header().to_owned());
-        registry.put_shared(dag).await?;
+
+        registry.register::<FlexidagService>().await?;
+
         info!(
             "Start node with chain info: {}, number {} upgrade_time cost {} secs, ",
             chain_info,

storage/src/lib.rs

@@ -26,6 +26,7 @@ use starcoin_accumulator::{
 };
 use starcoin_config::ChainNetworkID;
 use starcoin_crypto::HashValue;
+use starcoin_logger::prelude::info;
 use starcoin_state_store_api::{StateNode, StateNodeStore};
 use starcoin_types::block::BlockNumber;
 use starcoin_types::{
@@ -37,7 +38,7 @@ use starcoin_types::{
 };
 use starcoin_vm_types::{
     account_address::AccountAddress,
-    state_store::table::{TableHandle, TableInfo},
+    state_store::table::{TableHandle, TableInfo}, account_config::key_rotation_capability,
 };
 use std::{
     collections::{BTreeMap, BTreeSet},
@@ -453,15 +454,25 @@ impl BlockStore for Storage {
         let head_block_info = self.get_block_info(head_block.id())?.ok_or_else(|| {
             format_err!("Startup block info {:?} should exist", startup_info.main)
         })?;
-        let snapshot = self
+        let (tips, dag_accumulator_info, k_total_difficulties) = self
             .get_lastest_snapshot()?
-            .ok_or_else(|| anyhow!("latest snapshot is  none"))?;
+            .map_or_else(|| {
+                info!("the dag data is none, the chain will be still a single chain");
+                (None, None, None)
+            }, |snapshot| {
+                info!("the dag data exists, the chain will be still a dag chain");
+                (
+                    Some(snapshot.child_hashes),
+                    Some(snapshot.accumulator_info),
+                    Some(snapshot.k_total_difficulties),
+                )
+            });
         let chain_info = ChainInfo::new(
             head_block.chain_id(),
             genesis_hash,
-            ChainStatus::new(head_block, head_block_info,Some(snapshot.child_hashes)),
-            Some(snapshot.accumulator_info),
-            Some(snapshot.k_total_difficulties),
+            ChainStatus::new(head_block, head_block_info, tips),
+            dag_accumulator_info,
+            k_total_difficulties,
         );
         Ok(Some(chain_info))
     }
@@ -675,9 +686,11 @@ impl SyncFlexiDagStore for Storage {
     }
 
     fn get_lastest_snapshot(&self) -> Result<Option<SyncFlexiDagSnapshot>> {
-        let info = self
-            .get_dag_accumulator_info()?
-            .ok_or_else(|| anyhow!("dag startup info is none"))?;
+        let info = match self
+            .get_dag_accumulator_info()? {
+                Some(info) => info,
+                None => return Ok(None),
+            };
         let merkle_tree = MerkleAccumulator::new_with_info(
             info,
             self.get_accumulator_store(AccumulatorStoreType::SyncDag),

sync/src/tasks/block_sync_task.rs

@@ -3,7 +3,7 @@
 
 use crate::tasks::{BlockConnectedEventHandle, BlockFetcher, BlockLocalStore};
 use crate::verified_rpc_client::RpcVerifyError;
-use anyhow::{format_err, Ok, Result};
+use anyhow::{format_err, Ok, Result, anyhow};
 use futures::future::BoxFuture;
 use futures::FutureExt;
 use network_api::PeerId;
@@ -14,7 +14,7 @@ use starcoin_chain::{verifier::BasicVerifier, BlockChain};
 use starcoin_chain_api::{ChainReader, ChainWriter, ConnectBlockError, ExecutedBlock};
 use starcoin_config::G_CRATE_VERSION;
 use starcoin_crypto::HashValue;
-use starcoin_flexidag::flexidag_service::{FinishSync, ForkDagAccumulator};
+use starcoin_flexidag::flexidag_service::{FinishSync, ForkDagAccumulator, AddToDag};
 use starcoin_flexidag::FlexidagService;
 use starcoin_logger::prelude::*;
 use starcoin_service_registry::ServiceRef;
@@ -465,32 +465,34 @@ where
     }
 
     fn collect_dag_item(&mut self, item: SyncBlockData) -> Result<()> {
-        // let (block, block_info, peer_id) = item.into();
-        // let block_id = block.id();
-        // let timestamp = block.header().timestamp();
-
-        // let add_dag_result = async_std::task::block_on(self.flexidag_service.send(AddToDag {
-        //     block_header: block.header().clone(),
-        // }))??;
-        // let selected_parent = self
-        //     .storage
-        //     .get_block_by_hash(add_dag_result.selected_parent)?
-        //     .expect("selected parent should in storage");
-        // let mut chain = self.chain.fork(selected_parent.header.parent_hash())?;
-        // for blue_hash in add_dag_result.mergeset_blues.mergeset_blues.iter() {
-        //     if let Some(blue_block) = self.storage.get_block(blue_hash.to_owned())? {
-        //         match chain.apply(blue_block) {
-        //             std::result::Result::Ok(_executed_block) => (),
-        //             Err(e) => warn!("failed to connect dag block: {:?}", e),
-        //         }
-        //     } else {
-        //         error!("Failed to get block {:?}", blue_hash);
-        //     }
-        // }
-
-        // if chain.status().info().total_difficulty > self.chain.status().info().total_difficulty {
-        //     self.chain = chain;
-        // }
+        let (block, block_info, peer_id) = item.into();
+        let block_id = block.id();
+        let timestamp = block.header().timestamp();
+
+        let add_dag_result = self.flexidag_service.as_ref().map(|service| {
+            async_std::task::block_on(service.send(AddToDag {
+                        block_header: block.header().clone(),
+            }))? 
+        }).ok_or_else(|| anyhow!("flexidag service is None"))??;
+        let selected_parent = self
+            .storage
+            .get_block_by_hash(add_dag_result.selected_parent)?
+            .expect("selected parent should in storage");
+        let mut chain = self.chain.fork(selected_parent.header.parent_hash())?;
+        for blue_hash in add_dag_result.mergeset_blues.iter() {
+            if let Some(blue_block) = self.storage.get_block(blue_hash.to_owned())? {
+                match chain.apply(blue_block) {
+                    std::result::Result::Ok(_executed_block) => (),
+                    Err(e) => warn!("failed to connect dag block: {:?}", e),
+                }
+            } else {
+                error!("Failed to get block {:?}", blue_hash);
+            }
+        }
+
+        if chain.status().info().total_difficulty > self.chain.status().info().total_difficulty {
+            self.chain = chain;
+        }
 
         Ok(())
     }

types/src/block.rs

@@ -347,7 +347,7 @@ impl BlockHeader {
             chain_id,
             0,
             BlockHeaderExtra::default(),
-            None,
+            Some(vec![]),
         )
     }