Fix bug: no store the sync dag block

flexidag/src/blockdag.rs

@@ -226,7 +226,13 @@ impl BlockDAG {
         let mut merge_set = ghostdata
             .unordered_mergeset_without_selected_parent()
             .filter(|hash| self.storage.reachability_store.read().has(*hash).unwrap())
-            .collect::<Vec<_>>()
+            .chain(
+                header
+                    .parents_hash()
+                    .into_iter()
+                    .filter(|parent_id| *parent_id != ghostdata.selected_parent),
+            )
+            .collect::<HashSet<_>>()
             .into_iter();
         let add_block_result = {
             let mut reachability_writer = reachability_store.write();
@@ -482,8 +488,8 @@ impl BlockDAG {
         let dag_state = self.get_dag_state(previous_pruning_point)?;
         let next_ghostdata = self.ghostdata(&dag_state.tips)?;
         info!(
-            "start to calculate the mergeset and tips for tips: {:?}, and last pruning point: {:?} and next ghostdata: {:?}",
-            dag_state.tips, previous_pruning_point, next_ghostdata,
+            "start to calculate the mergeset and tips for tips: {:?}, and last pruning point: {:?} and next ghostdata's selected parents: {:?} and blues set are {:?}",
+            dag_state.tips, previous_pruning_point, next_ghostdata.selected_parent, next_ghostdata.mergeset_blues,
         );
         let next_pruning_point = self.pruning_point_manager().next_pruning_point(
             previous_pruning_point,

flexidag/src/ghostdag/protocol.rs

@@ -210,22 +210,29 @@ impl<
             }
         }
 
+        let remote_blue_set = blue_blocks
+            .iter()
+            .map(|header| header.id())
+            .collect::<HashSet<_>>();
         if new_block_data
             .mergeset_blues
             .iter()
             .skip(1)
             .cloned()
             .collect::<HashSet<_>>()
-            != blue_blocks
+            != remote_blue_set
+        {
+            warn!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
+            let ghostdata = self.ghostdag(&header.parents_hash())?;
+            if ghostdata
+                .mergeset_blues
                 .iter()
-                .map(|header| header.id())
+                .skip(1)
+                .cloned()
                 .collect::<HashSet<_>>()
-        {
-            if header.number() < 10000000 {
-                // no bail before 10000000
-                warn!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
-            } else {
-                bail!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
+                != remote_blue_set
+            {
+                bail!("The ghost data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), ghostdata.mergeset_blues);
             }
         }
 
@@ -377,12 +384,20 @@ impl<
             *candidate_blue_anticone_size = (*candidate_blue_anticone_size).checked_add(1).unwrap();
             if *candidate_blue_anticone_size > self.k {
                 // k-cluster violation: The candidate's blue anticone exceeded k
+                info!(
+                    "Checking blue candidate: {} failed, blue anticone exceeded k",
+                    blue_candidate
+                );
                 return Ok(ColoringState::Red);
             }
 
             if *candidate_blues_anticone_sizes.get(&block).unwrap() == self.k {
                 // k-cluster violation: A block in candidate's blue anticone already
                 // has k blue blocks in its own anticone
+                info!(
+                    "Checking blue candidate: {} failed, block {} has k blue blocks in its anticone",
+                    blue_candidate, block
+                );
                 return Ok(ColoringState::Red);
             }
 
@@ -431,6 +446,10 @@ impl<
         // The maximum length of new_block_data.mergeset_blues can be K+1 because
         // it contains the selected parent.
         if new_block_data.mergeset_blues.len() as KType == self.k.checked_add(1).unwrap() {
+            info!(
+                "Checking blue candidate: {} failed, mergeset blues size is K+1",
+                blue_candidate
+            );
             return Ok(ColoringOutput::Red);
         }
 

flexidag/src/prune/pruning_point_manager.rs

@@ -75,6 +75,11 @@ impl<T: ReachabilityStoreReader + Clone> PruningPointManagerT<T> {
             min_required_blue_score_for_next_pruning_point
         );
 
+        debug!("previous_pruning_point: {:?}, previous_ghostdata: {:?}, next_ghostdata: {:?}, pruning_depth: {:?}, pruning_finality: {:?}", 
+            previous_pruning_point, previous_ghostdata, next_ghostdata,
+            pruning_depth, pruning_finality,
+        );
+
         let mut latest_pruning_ghost_data = previous_ghostdata.to_compact();
         if min_required_blue_score_for_next_pruning_point + pruning_depth
             <= next_ghostdata.blue_score

flexidag/tests/tests.rs

@@ -19,10 +19,12 @@ use starcoin_logger::prelude::debug;
 use starcoin_types::{
     block::{BlockHeader, BlockHeaderBuilder, BlockNumber},
     blockhash::{BlockHashMap, HashKTypeMap, KType},
+    consensus_header::ConsensusHeader,
+    U256,
 };
 
 use std::{
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     ops::{Deref, DerefMut},
     sync::Arc,
     time::Instant,
@@ -737,6 +739,31 @@ fn add_and_print_with_ghostdata(
     Ok(header)
 }
 
+fn add_and_print_with_difficulty(
+    number: BlockNumber,
+    parent: Hash,
+    parents: Vec<Hash>,
+    difficulty: U256,
+) -> anyhow::Result<BlockHeader> {
+    let header_builder = BlockHeaderBuilder::random();
+    let header = header_builder
+        .with_parent_hash(parent)
+        .with_parents_hash(parents)
+        .with_number(number)
+        .with_difficulty(difficulty)
+        .build();
+    let start = Instant::now();
+    let duration = start.elapsed();
+    println!(
+        "commit header: {:?}, number: {:?}, duration: {:?}",
+        header.id(),
+        header.number(),
+        duration
+    );
+
+    Ok(header)
+}
+
 fn add_and_print_with_pruning_point(
     number: BlockNumber,
     parent: Hash,
@@ -751,6 +778,7 @@ fn add_and_print_with_pruning_point(
         .with_parents_hash(parents)
         .with_number(number)
         .with_pruning_point(pruning_point)
+        .with_difficulty(U256::from(10))
         .build();
     let start = Instant::now();
     dag.commit(header.to_owned(), origin)?;
@@ -761,10 +789,11 @@ fn add_and_print_with_pruning_point(
         header.number(),
         duration
     );
-    let _ghostdata = dag.ghostdata(&[header.id()])?;
+    // let ghostdata = dag.ghostdata(&[header.id()])?;
+    // let ghostdata = dag.ghostdata_by_hash(header.id())?.unwrap();
     // println!(
-    //     "add a header: {:?}, blue set: {:?}, red set: {:?}, blue anticone size: {:?}",
-    //     header, ghostdata.mergeset_blues, ghostdata.mergeset_reds, ghostdata.blues_anticone_sizes
+    //     "add a header: {:?}, selected_parent: {:?}, blue set: {:?}, red set: {:?}, blue anticone size: {:?}",
+    //     header, ghostdata.selected_parent, ghostdata.mergeset_blues, ghostdata.mergeset_reds, ghostdata.blues_anticone_sizes
     // );
     Ok(header)
 }
@@ -1069,6 +1098,193 @@ fn test_prune() -> anyhow::Result<()> {
     anyhow::Result::Ok(())
 }
 
+#[test]
+fn test_verification_blue_block_inconsistent() -> anyhow::Result<()> {
+    loop_to_blue()?;
+    anyhow::Result::Ok(())
+}
+
+fn loop_to_blue() -> anyhow::Result<()> {
+    // initialzie the dag firstly
+    let k = 2;
+
+    let mut dag = BlockDAG::create_for_testing_with_parameters(k).unwrap();
+
+    let origin = BlockHeaderBuilder::random().with_number(0).build();
+    let genesis = BlockHeader::dag_genesis_random_with_parent(origin)?;
+
+    dag.init_with_genesis(genesis.clone()).unwrap();
+
+    let mut storage = HashMap::new();
+
+    let block1 =
+        add_and_print_with_difficulty(1, genesis.id(), vec![genesis.id()], U256::from(10))?;
+    storage.insert(block1.id(), block1.clone());
+    let ghost = dag.ghostdata(&block1.parents())?;
+    let verified_ghost = dag.verify_and_ghostdata(
+        &ghost
+            .mergeset_blues
+            .iter()
+            .skip(1)
+            .cloned()
+            .map(|x| storage.get(&x).unwrap().clone())
+            .collect::<Vec<_>>(),
+        &block1,
+    )?;
+    dag.commit_trusted_block(
+        block1.clone(),
+        genesis.parent_hash(),
+        Arc::new(verified_ghost),
+    )?;
+
+    let mut bottom = vec![];
+    let mut last = block1.clone();
+    for i in 0..500 {
+        let block2 =
+            add_and_print_with_difficulty(1 + i, last.id(), vec![last.id()], U256::from(10))?;
+        last = block2.clone();
+        storage.insert(block2.id(), block2.clone());
+        let ghost = dag.ghostdata(&block2.parents())?;
+        let verified_ghost = dag.verify_and_ghostdata(
+            &ghost
+                .mergeset_blues
+                .iter()
+                .skip(1)
+                .cloned()
+                .map(|x| storage.get(&x).unwrap().clone())
+                .collect::<Vec<_>>(),
+            &block2,
+        )?;
+        dag.commit_trusted_block(
+            block2.clone(),
+            genesis.parent_hash(),
+            Arc::new(verified_ghost),
+        )?;
+        bottom.push(block2);
+    }
+
+    let mut top = vec![];
+    let mut iter = bottom.iter().peekable();
+    while let Some(first) = iter.next() {
+        if let Some(second) = iter.next() {
+            let block = add_and_print_with_difficulty(
+                3,
+                first.id(),
+                vec![first.id(), second.id()],
+                U256::from(10),
+            )?;
+            storage.insert(block.id(), block.clone());
+            let ghost = dag.ghostdata(&block.parents())?;
+            let verified_ghost = dag.verify_and_ghostdata(
+                &ghost
+                    .mergeset_blues
+                    .iter()
+                    .skip(1)
+                    .cloned()
+                    .map(|x| storage.get(&x).unwrap().clone())
+                    .collect::<Vec<_>>(),
+                &block,
+            )?;
+            dag.commit_trusted_block(
+                block.clone(),
+                genesis.parent_hash(),
+                Arc::new(verified_ghost),
+            )?;
+
+            last = block.clone();
+            top.push(block);
+        } else {
+            let block = add_and_print_with_difficulty(
+                3,
+                first.id(),
+                vec![first.id(), last.id()],
+                U256::from(10),
+            )?;
+            storage.insert(block.id(), block.clone());
+            let ghost = dag.ghostdata(&block.parents())?;
+            let verified_ghost = dag.verify_and_ghostdata(
+                &ghost
+                    .mergeset_blues
+                    .iter()
+                    .skip(1)
+                    .cloned()
+                    .map(|x| storage.get(&x).unwrap().clone())
+                    .collect::<Vec<_>>(),
+                &block,
+            )?;
+            dag.commit_trusted_block(
+                block.clone(),
+                genesis.parent_hash(),
+                Arc::new(verified_ghost),
+            )?;
+
+            top.push(block);
+            if top.len() == 1 {
+                last = top[0].clone();
+                break;
+            } else {
+                bottom.clone_from(&top);
+                iter = bottom.iter().peekable();
+                top.clear();
+            }
+        }
+    }
+
+    let block1_1 = add_and_print_with_difficulty(
+        1,
+        genesis.id(),
+        vec![last.id(), block1.id()],
+        U256::from(99999999),
+    )?;
+    storage.insert(block1_1.id(), block1_1.clone());
+    let ghost = dag.ghostdata(&block1_1.parents())?;
+    let verified_ghost = dag.verify_and_ghostdata(
+        &ghost
+            .mergeset_blues
+            .iter()
+            .skip(1)
+            .cloned()
+            .map(|x| storage.get(&x).unwrap().clone())
+            .collect::<Vec<_>>(),
+        &block1_1,
+    )?;
+    dag.commit_trusted_block(
+        block1_1.clone(),
+        genesis.parent_hash(),
+        Arc::new(verified_ghost),
+    )?;
+
+    let block3 = add_and_print_with_difficulty(
+        3,
+        block1_1.id(),
+        vec![block1_1.id(), last.id()],
+        U256::from(10),
+    )?;
+
+    let ghostdata = dag.ghostdata(&block3.parents())?;
+    println!(
+        "add a header: {:?}, selected_parent: {:?}, blue set: {:?}, red set: {:?}, blue anticone size: {:?}",
+        block3, ghostdata.selected_parent, ghostdata.mergeset_blues, ghostdata.mergeset_reds, ghostdata.blues_anticone_sizes
+    );
+    let verified_ghostdata = dag.verify_and_ghostdata(
+        &ghostdata
+            .mergeset_blues
+            .iter()
+            .skip(1)
+            .map(|x| dag.storage.header_store.get_header(*x).unwrap())
+            .collect::<Vec<_>>(),
+        &block3,
+    )?;
+    println!(
+        "after verification: selected_parent: {:?}, blue set: {:?}, red set: {:?}, blue anticone size: {:?}",
+        verified_ghostdata.selected_parent, verified_ghostdata.mergeset_blues, verified_ghostdata.mergeset_reds, verified_ghostdata.blues_anticone_sizes
+    );
+
+    assert_eq!(ghostdata.mergeset_blues, verified_ghostdata.mergeset_blues);
+
+    anyhow::Ok(())
+}
+
 #[test]
 fn test_verification_blue_block() -> anyhow::Result<()> {
     // initialzie the dag firstly

kube/manifest/starcoin-vega.yaml

@@ -28,7 +28,7 @@ spec:
           - -c
         args:
           -
-            rm -rf /sc-data/vega/starcoin.ipc /sc-data/vega/starcoindb/db/starcoindb/LOCK /sc-data/vega/genesis_config.json;
+            rm -rf /sc-data/vega/sync /sc-data/vega/starcoin.ipc /sc-data/vega/starcoindb/db/starcoindb/LOCK /sc-data/vega/genesis_config.json;
             id=$(echo -e $POD_NAME|awk -F'-' '{print $2}') && IFS='; ' read -r -a node_keys <<< $NODE_KEYS &&
             node_key=${node_keys[$id]};
             if [ ! -z $node_key ]; then

sync/src/tasks/block_sync_task.rs

@@ -454,6 +454,7 @@ where
             if block_header.number() % ASYNC_BLOCK_COUNT == 0
                 || block_header.number() >= self.target.target_id.number()
             {
+                self.sync_dag_store.delete_all_dag_sync_block()?;
                 self.find_absent_ancestor(vec![block_header.clone()])
                     .await?;
 
@@ -474,7 +475,6 @@ where
                         block: block.clone(),
                         children: vec![],
                     })?;
-                self.sync_dag_store.save_block(block)?;
                 anyhow::Ok(ParallelSign::NeedMoreBlocks)
             }
         };