add basic voting tests

consensus/core/src/block.rs

@@ -682,13 +682,13 @@ impl BlockOutput {
 /// This struct is public for testing in other crates.
 #[derive(Clone)]
 pub struct TestBlock {
-    block: BlockV1,
+    block: BlockV2,
 }
 
 impl TestBlock {
     pub fn new(round: Round, author: u32) -> Self {
         Self {
-            block: BlockV1 {
+            block: BlockV2 {
                 round,
                 author: AuthorityIndex::new_for_test(author),
                 ..Default::default()
@@ -726,13 +726,20 @@ impl TestBlock {
         self
     }
 
-    pub fn set_commit_votes(mut self, commit_votes: Vec<CommitVote>) -> Self {
+    #[cfg(test)]
+    pub(crate) fn set_transaction_votes(mut self, votes: Vec<BlockTransactionVotes>) -> Self {
+        self.block.transaction_votes = votes;
+        self
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_commit_votes(mut self, commit_votes: Vec<CommitVote>) -> Self {
         self.block.commit_votes = commit_votes;
         self
     }
 
     pub fn build(self) -> Block {
-        Block::V1(self.block)
+        Block::V2(self.block)
     }
 }
 

consensus/core/src/dag_state.rs

@@ -1137,7 +1137,10 @@ mod test {
 
     use super::*;
     use crate::{
-        block::{BlockDigest, BlockRef, BlockTimestampMs, TestBlock, VerifiedBlock},
+        block::{
+            BlockDigest, BlockRef, BlockTimestampMs, BlockTransactionVotes, TestBlock,
+            VerifiedBlock,
+        },
         storage::{mem_store::MemStore, WriteBatch},
         test_dag_builder::DagBuilder,
         test_dag_parser::parse_dag,
@@ -1701,6 +1704,170 @@ mod test {
         assert_eq!(result, expected);
     }
 
+    #[tokio::test]
+    async fn test_voting_basic() {
+        telemetry_subscribers::init_for_testing();
+        let num_authorities: u32 = 7;
+        let (context, _) = Context::new_for_test(num_authorities as usize);
+        let context = Arc::new(context);
+        let store = Arc::new(MemStore::new());
+        let mut dag_state = DagState::new(context.clone(), store.clone());
+
+        // Create minimal connected blocks up to round voting_rounds - 1,
+        // and add a final round with full blocks connections.
+        let voting_rounds = context.protocol_config.consensus_voting_rounds();
+        let num_rounds = voting_rounds - 1;
+        let mut dag_builder = DagBuilder::new(context.clone());
+        dag_builder
+            .layers(1..=num_rounds)
+            .min_ancestor_links(false, None);
+        dag_builder.layer(voting_rounds).build();
+
+        // Add all created blocks to DagState.
+        let mut all_blocks: Vec<_> = dag_builder.all_blocks();
+        all_blocks.sort_by_key(|b| b.reference());
+        dag_state.accept_blocks(all_blocks.clone());
+
+        let certified_blocks = dag_state.take_certified_blocks();
+
+        // It is expected that all blocks with round < voting_rounds are certified.
+        let voted_block_refs = all_blocks
+            .iter()
+            .filter_map(|b| {
+                if b.round() < voting_rounds {
+                    Some(b.reference())
+                } else {
+                    None
+                }
+            })
+            .collect::<BTreeSet<_>>();
+        let certified_block_refs = certified_blocks
+            .iter()
+            .map(|b| b.block.reference())
+            .collect::<BTreeSet<_>>();
+
+        let diff = voted_block_refs
+            .difference(&certified_block_refs)
+            .collect::<Vec<_>>();
+        assert!(diff.is_empty(), "Blocks {:?} are not certified", diff);
+
+        let diff = certified_block_refs
+            .difference(&voted_block_refs)
+            .collect::<Vec<_>>();
+        assert!(
+            diff.is_empty(),
+            "Certified blocks {:?} are unexpected",
+            diff
+        );
+
+        // Ensure no transaction is rejected.
+        for b in &certified_blocks {
+            assert!(b.rejected.is_empty());
+        }
+    }
+
+    #[tokio::test]
+    async fn test_voting_with_rejections() {
+        telemetry_subscribers::init_for_testing();
+        let num_authorities: u32 = 4;
+        let (context, _) = Context::new_for_test(num_authorities as usize);
+        let context = Arc::new(context);
+        let store = Arc::new(MemStore::new());
+        let mut dag_state = DagState::new(context.clone(), store.clone());
+
+        // Create connected blocks up to voting_rounds, with only 3 authorities.
+        let voting_rounds = context.protocol_config.consensus_voting_rounds();
+        let last_round = voting_rounds + 1;
+        let mut dag_builder = DagBuilder::new(context.clone());
+        dag_builder
+            .layers(1..=last_round)
+            .authorities((0..3).map(AuthorityIndex::new_for_test).collect())
+            .include_transactions(4)
+            .build();
+
+        let mut all_blocks: Vec<_> = dag_builder.all_blocks();
+        all_blocks.sort_by_key(|b| b.reference());
+
+        let last_block = all_blocks.last().unwrap().clone();
+        assert_eq!(last_block.round(), last_round);
+
+        let mut next_ancestors = all_blocks
+            .iter()
+            .filter_map(|b| {
+                if b.round() == last_round {
+                    Some(b.reference())
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<_>>();
+
+        // Create a block outside of voting rounds.
+        let round_1_block = VerifiedBlock::new_for_test(TestBlock::new(1, 3).build());
+        next_ancestors.push(round_1_block.reference());
+
+        // Create blocks with rejection votes.
+        let final_round_blocks: Vec<_> = (0..4)
+            .map(|i| {
+                let test_block = TestBlock::new(last_round + 1, i)
+                    .set_transaction_votes(vec![BlockTransactionVotes {
+                        block_ref: last_block.reference(),
+                        rejects: vec![2],
+                    }])
+                    .set_ancestors(next_ancestors.clone())
+                    .build();
+                VerifiedBlock::new_for_test(test_block)
+            })
+            .collect();
+
+        // Accept all created blocks.
+        all_blocks.push(round_1_block);
+        all_blocks.extend(final_round_blocks);
+        dag_state.accept_blocks(all_blocks.clone());
+
+        let certified_blocks = dag_state.take_certified_blocks();
+
+        // It is expected that all blocks with round <= last_round and from authorities [0,1,2] are certified.
+        // The rest of blocks are not.
+        let voted_block_refs = all_blocks
+            .iter()
+            .filter_map(|b| {
+                if b.round() <= last_round && b.author() != AuthorityIndex::new_for_test(3) {
+                    Some(b.reference())
+                } else {
+                    None
+                }
+            })
+            .collect::<BTreeSet<_>>();
+        let certified_block_refs = certified_blocks
+            .iter()
+            .map(|b| b.block.reference())
+            .collect::<BTreeSet<_>>();
+
+        let diff = voted_block_refs
+            .difference(&certified_block_refs)
+            .collect::<Vec<_>>();
+        assert!(diff.is_empty(), "Blocks {:?} are not certified", diff);
+
+        let diff = certified_block_refs
+            .difference(&voted_block_refs)
+            .collect::<Vec<_>>();
+        assert!(
+            diff.is_empty(),
+            "Certified blocks {:?} are unexpected",
+            diff
+        );
+
+        // Ensure only the expected transaction is rejected.
+        for b in &certified_blocks {
+            if b.block.reference() != last_block.reference() {
+                assert!(b.rejected.is_empty());
+                continue;
+            }
+            assert_eq!(b.rejected, vec![2]);
+        }
+    }
+
     // TODO: Remove when DistributedVoteScoring is enabled.
     #[rstest]
     #[tokio::test]

consensus/core/src/test_dag_builder.rs

@@ -2,7 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::{
-    collections::{BTreeMap, HashSet},
+    collections::{BTreeMap, BTreeSet},
+    iter,
     ops::{Bound::Included, RangeInclusive},
     sync::Arc,
 };
@@ -21,7 +22,7 @@ use crate::{
     dag_state::DagState,
     leader_schedule::{LeaderSchedule, LeaderSwapTable},
     linearizer::{BlockStoreAPI, Linearizer},
-    CommittedSubDag,
+    CommittedSubDag, Transaction,
 };
 
 /// DagBuilder API
@@ -379,6 +380,8 @@ pub struct LayerBuilder<'a> {
     // Configuration options applied to specified authorities
     // TODO: convert configuration options into an enum
     specified_authorities: Option<Vec<AuthorityIndex>>,
+    // Number of transactions to include per block.
+    num_transactions: Option<u32>,
     // Number of equivocating blocks per specified authority
     equivocations: usize,
     // Skip block proposal for specified authorities
@@ -422,6 +425,7 @@ impl<'a> LayerBuilder<'a> {
             start_round,
             end_round: None,
             specified_authorities: None,
+            num_transactions: None,
             equivocations: 0,
             skip_block: false,
             skip_ancestor_links: None,
@@ -507,6 +511,11 @@ impl<'a> LayerBuilder<'a> {
         self
     }
 
+    pub fn include_transactions(mut self, num_transactions: u32) -> Self {
+        self.num_transactions = Some(num_transactions);
+        self
+    }
+
     // Multiple blocks will be created for the specified authorities at the layer round.
     pub fn equivocate(mut self, equivocations: usize) -> Self {
         // authorities must be specified for this to apply
@@ -542,9 +551,9 @@ impl<'a> LayerBuilder<'a> {
             // TODO: investigate if these configurations can be called in combination
             // for the same layer
             let mut connections = if self.fully_linked_ancestors {
-                self.configure_fully_linked_ancestors()
+                self.configure_fully_linked_ancestors(authorities)
             } else if self.min_ancestor_links {
-                self.configure_min_parent_links()
+                self.configure_min_parent_links(authorities)
             } else if self.no_leader_link {
                 self.configure_no_leader_links(authorities.clone(), round)
             } else if self.skip_ancestor_links.is_some() {
@@ -573,23 +582,17 @@ impl<'a> LayerBuilder<'a> {
     }
 
     // Layer round is minimally and randomly connected with ancestors.
-    pub fn configure_min_parent_links(&mut self) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
+    pub fn configure_min_parent_links(
+        &mut self,
+        authorities: Vec<AuthorityIndex>,
+    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
         let quorum_threshold = self.dag_builder.context.committee.quorum_threshold() as usize;
-        let mut authorities: Vec<AuthorityIndex> = self
-            .dag_builder
-            .context
-            .committee
-            .authorities()
-            .map(|authority| authority.0)
-            .collect();
 
         let mut rng = match self.min_ancestor_links_random_seed {
             Some(s) => StdRng::seed_from_u64(s),
             None => StdRng::from_entropy(),
         };
 
-        let mut authorities_to_shuffle = authorities.clone();
-
         let mut leaders = vec![];
         if let Some(leader_round) = self.leader_round {
             let leader_offsets = (0..self.dag_builder.number_of_leaders).collect::<Vec<_>>();
@@ -603,27 +606,37 @@ impl<'a> LayerBuilder<'a> {
             }
         }
 
+        let mut authorities_to_shuffle = authorities.clone();
+
         authorities
             .iter()
             .map(|authority| {
                 authorities_to_shuffle.shuffle(&mut rng);
 
-                // TODO: handle quroum threshold properly with stake
-                let min_ancestors: HashSet<AuthorityIndex> = authorities_to_shuffle
+                // TODO: handle quorum threshold properly with stake
+                let min_ancestors: BTreeSet<AuthorityIndex> = authorities_to_shuffle
                     .iter()
+                    .filter(|a| authority != *a)
                     .take(quorum_threshold)
                     .cloned()
                     .collect();
 
                 (
                     *authority,
-                    self.ancestors
-                        .iter()
-                        .filter(|a| {
-                            leaders.contains(&a.author) || min_ancestors.contains(&a.author)
-                        })
-                        .cloned()
-                        .collect::<Vec<BlockRef>>(),
+                    // Make sure the authority ancestor is the 1st.
+                    // And it is not given that the authority ancestor is a parent, so it is still necessary
+                    // to have 2f+1 other ancestors.
+                    iter::once(
+                        self.ancestors
+                            .iter()
+                            .find(|a| *authority == a.author)
+                            .unwrap(),
+                    )
+                    .chain(self.ancestors.iter().filter(|a| {
+                        leaders.contains(&a.author) || min_ancestors.contains(&a.author)
+                    }))
+                    .cloned()
+                    .collect::<Vec<BlockRef>>(),
                 )
             })
             .collect()
@@ -664,12 +677,13 @@ impl<'a> LayerBuilder<'a> {
         self.configure_skipped_ancestor_links(authorities, missing_leaders)
     }
 
-    fn configure_fully_linked_ancestors(&mut self) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
-        self.dag_builder
-            .context
-            .committee
-            .authorities()
-            .map(|authority| (authority.0, self.ancestors.clone()))
+    fn configure_fully_linked_ancestors(
+        &mut self,
+        authorities: Vec<AuthorityIndex>,
+    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
+        authorities
+            .into_iter()
+            .map(|authority| (authority, self.ancestors.clone()))
             .collect::<Vec<_>>()
     }
 
@@ -703,12 +717,16 @@ impl<'a> LayerBuilder<'a> {
             for num_block in 0..num_blocks {
                 let author = authority.value() as u32;
                 let base_ts = round as BlockTimestampMs * 1000;
-                let block = VerifiedBlock::new_for_test(
-                    TestBlock::new(round, author)
-                        .set_ancestors(ancestors.clone())
-                        .set_timestamp_ms(base_ts + (author + round + num_block) as u64)
-                        .build(),
-                );
+                let mut test_bock = TestBlock::new(round, author)
+                    .set_ancestors(ancestors.clone())
+                    .set_timestamp_ms(base_ts + (author + round + num_block) as u64);
+                if let Some(num_transactions) = self.num_transactions {
+                    let transactions = (0..num_transactions)
+                        .map(|_| Transaction::new(vec![0_u8; 16]))
+                        .collect();
+                    test_bock = test_bock.set_transactions(transactions);
+                };
+                let block = VerifiedBlock::new_for_test(test_bock.build());
                 references.push(block.reference());
                 self.dag_builder
                     .blocks