add sync finish logic

Author: jackzhhuang

flexidag/src/flexidag_service.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use anyhow::{anyhow, bail, Error, Ok, Result};
-use starcoin_accumulator::{accumulator_info::AccumulatorInfo, Accumulator, MerkleAccumulator};
+use starcoin_accumulator::{accumulator_info::AccumulatorInfo, Accumulator, MerkleAccumulator, node::AccumulatorStoreType};
 use starcoin_config::{NodeConfig, TimeService};
 use starcoin_consensus::{dag::types::ghostdata::GhostdagData, BlockDAG};
 use starcoin_crypto::HashValue;
@@ -14,7 +14,7 @@ use starcoin_service_registry::{
 use starcoin_storage::{
     flexi_dag::{KTotalDifficulty, SyncFlexiDagSnapshot, SyncFlexiDagSnapshotHasher},
     storage::CodecKVStore,
-    BlockStore, Storage, SyncFlexiDagStore, block_info::BlockInfoStore,
+    BlockStore, Storage, SyncFlexiDagStore, block_info::BlockInfoStore, Store,
 };
 use starcoin_types::{block::BlockHeader, header::DagHeader, startup_info};
 
@@ -127,6 +127,15 @@ impl ServiceRequest for ForkDagAccumulator {
     type Response = anyhow::Result<AccumulatorInfo>;
 }
 
+#[derive(Debug, Clone)]
+pub struct FinishSync {
+    pub dag_accumulator_info: AccumulatorInfo,
+}
+
+impl ServiceRequest for FinishSync {
+    type Response = anyhow::Result<()>;
+}
+
 pub struct TipInfo {
     tips: Option<Vec<HashValue>>, // some is for dag or the state of the chain is still in old version
     k_total_difficulties: BTreeSet<KTotalDifficulty>,
@@ -536,8 +545,28 @@ impl ServiceHandler<Self, ForkDagAccumulator> for FlexidagService {
         } else {
             self.merge_from_small_dag(msg)
         }
-
-
-        // append the ForkDagAccumulator.new_blocks and the fetched blocks above into the forked dag accumulator
     }
 }
+
+impl ServiceHandler<Self, FinishSync> for FlexidagService {
+    fn handle(
+        &mut self,
+        msg: FinishSync,
+        _ctx: &mut ServiceContext<FlexidagService>,
+    ) -> Result<()> {
+        let dag_accumulator = self.dag_accumulator.ok_or_else(|| anyhow!("the dag_accumulator is none when sync finish"))?;
+        let local_info = dag_accumulator.get_info();
+        if msg.dag_accumulator_info.get_num_leaves() < local_info.get_num_leaves() {
+            let mut new_dag_accumulator = MerkleAccumulator::new_with_info(msg.dag_accumulator_info, self.storage.get_accumulator_store(AccumulatorStoreType::SyncDag));
+            for index in msg.dag_accumulator_info.get_num_leaves()..local_info.get_num_leaves() {
+                let key = dag_accumulator.get_leaf(index)?.ok_or_else(|| anyhow!("the dag_accumulator leaf is none when sync finish"))?;
+                new_dag_accumulator.append(&[key])?;
+            }
+            self.dag_accumulator = Some(new_dag_accumulator);
+            Ok(())
+        } else {
+            self.dag_accumulator = Some(MerkleAccumulator::new_with_info(msg.dag_accumulator_info, self.storage.get_accumulator_store(AccumulatorStoreType::SyncDag)));
+            Ok(())
+        }
+    }
+}

sync/src/tasks/block_sync_task.rs

@@ -15,7 +15,7 @@ use starcoin_chain_api::{ChainReader, ChainWriter, ConnectBlockError, ExecutedBl
 use starcoin_config::{Connect, G_CRATE_VERSION};
 use starcoin_consensus::BlockDAG;
 use starcoin_crypto::HashValue;
-use starcoin_flexidag::flexidag_service::{AddToDag, GetDagTips, ForkDagAccumulator};
+use starcoin_flexidag::flexidag_service::{AddToDag, GetDagTips, ForkDagAccumulator, FinishSync};
 use starcoin_flexidag::FlexidagService;
 use starcoin_logger::prelude::*;
 use starcoin_service_registry::ServiceRef;
@@ -417,6 +417,11 @@ where
             dag_accumulator_index: start_index, 
             block_header_id: self.chain.head_block().id(),
         }))??);
+        if state == State::Enough {
+            async_std::task::block_on(self.flexidag_service.send(FinishSync {
+                dag_accumulator_info: self.new_dag_accumulator_info.clone(),
+            }))??
+        }
         return Ok(state);
     }