[consensus] Add failed_authors to Block

We want to have on the ledger who were the proposers/leaders for the failed rounds.
We are going to do that by adding failed_authors into the block, and
having proposer populate it.

Test Plan: added some unit tests for this

RFC:
- validators need to validate proposal, where is the best place for that?
- not sure if BlockType.Proposal.failed_authors, let me know if there is a better place.
- not sure if I need to add serialization for that field, or is that automatic.
- not sure what special needs to be done for changes like this that make a breaking protocol change

Closes: aptos-labs#1481

config/src/config/consensus_config.rs

@@ -30,6 +30,7 @@ pub struct ConsensusConfig {
     // the period = (poll_count - 1) * 30ms
     pub quorum_store_poll_count: u64,
     pub intra_consensus_channel_buffer_size: usize,
+    pub max_failed_authors_to_store: usize,
 }
 
 impl Default for ConsensusConfig {
@@ -52,6 +53,7 @@ impl Default for ConsensusConfig {
             quorum_store_pull_timeout_ms: 1000,
             quorum_store_poll_count: 20,
             intra_consensus_channel_buffer_size: 10,
+            max_failed_authors_to_store: 10,
         }
     }
 }

consensus/consensus-types/src/block.rs

@@ -194,10 +194,12 @@ impl Block {
         timestamp_usecs: u64,
         quorum_cert: QuorumCert,
         validator_signer: &ValidatorSigner,
+        failed_authors: Vec<(Round, Author)>,
     ) -> Self {
         let block_data = BlockData::new_proposal(
             payload,
             validator_signer.author(),
+            failed_authors,
             round,
             timestamp_usecs,
             quorum_cert,
@@ -264,6 +266,32 @@ impl Block {
                 "Reconfiguration suffix should not carry payload"
             );
         }
+        if let Some(failed_authors) = self.block_data().failed_authors() {
+            // when validating for being well formed,
+            // allow for missing failed authors,
+            // for whatever reason (from different max configuration, etc),
+            // but don't allow anything that shouldn't be there.
+            //
+            // we validate the full correctness of this field in round_manager.process_proposal()
+            let skipped_rounds = self.round().checked_sub(parent.round() + 1);
+            ensure!(
+                skipped_rounds.is_some(),
+                "Block round is smaller than block's parent round"
+            );
+            ensure!(
+                failed_authors.len() <= skipped_rounds.unwrap() as usize,
+                "Block has more failed authors than missed rounds"
+            );
+            let mut bound = parent.round();
+            for (round, _) in failed_authors {
+                ensure!(
+                    bound < *round && *round < self.round(),
+                    "Incorrect round in failed authors"
+                );
+                bound = *round;
+            }
+        }
+
         if self.is_nil_block() || parent.has_reconfiguration() {
             ensure!(
                 self.timestamp_usecs() == parent.timestamp_usecs(),

consensus/consensus-types/src/block_data.rs

@@ -23,6 +23,10 @@ pub enum BlockType {
         payload: Payload,
         /// Author of the block that can be validated by the author's public key and the signature
         author: Author,
+        /// Failed authors from the parent's block to this block.
+        /// I.e. the list of consecutive proposers from the
+        /// immediately preceeding rounds that didn't produce a successful block.
+        failed_authors: Vec<(Round, Author)>,
     },
     /// NIL blocks don't have authors or signatures: they're generated upon timeouts to fill in the
     /// gaps in the rounds.
@@ -116,6 +120,19 @@ impl BlockData {
         matches!(self.block_type, BlockType::NilBlock)
     }
 
+    /// the list of consecutive proposers from the immediately preceeding
+    /// rounds that didn't produce a successful block
+    pub fn failed_authors(&self) -> Option<&Vec<(Round, Author)>> {
+        if let BlockType::Proposal {
+            ref failed_authors, ..
+        } = self.block_type
+        {
+            Some(failed_authors)
+        } else {
+            None
+        }
+    }
+
     pub fn new_genesis_from_ledger_info(ledger_info: &LedgerInfo) -> Self {
         assert!(ledger_info.ends_epoch());
         let ancestor = BlockInfo::new(
@@ -188,6 +205,7 @@ impl BlockData {
     pub fn new_proposal(
         payload: Payload,
         author: Author,
+        failed_authors: Vec<(Round, Author)>,
         round: Round,
         timestamp_usecs: u64,
         quorum_cert: QuorumCert,
@@ -197,7 +215,11 @@ impl BlockData {
             round,
             timestamp_usecs,
             quorum_cert,
-            block_type: BlockType::Proposal { payload, author },
+            block_type: BlockType::Proposal {
+                payload,
+                author,
+                failed_authors,
+            },
         }
     }
 
@@ -235,6 +257,7 @@ fn test_reconfiguration_suffix() {
     let reconfig_suffix_block = BlockData::new_proposal(
         Payload::new_empty(),
         AccountAddress::random(),
+        Vec::new(),
         2,
         2,
         quorum_cert,

consensus/consensus-types/src/block_test.rs

@@ -6,14 +6,14 @@ use crate::{
         block_test_utils::{certificate_for_genesis, *},
         Block,
     },
-    common::Payload,
+    common::{Author, Payload},
     quorum_cert::QuorumCert,
     vote_data::VoteData,
 };
 use aptos_crypto::{hash::HashValue, test_utils::TestAptosCrypto};
 use aptos_types::{
     account_address::AccountAddress,
-    block_info::BlockInfo,
+    block_info::{BlockInfo, Round},
     ledger_info::{LedgerInfo, LedgerInfoWithSignatures},
     on_chain_config::ValidatorSet,
     validator_signer::ValidatorSigner,
@@ -74,6 +74,7 @@ fn test_nil_block() {
         aptos_infallible::duration_since_epoch().as_micros() as u64,
         nil_block_qc,
         &signer,
+        Vec::new(),
     );
     assert_eq!(nil_block_child.is_nil_block(), false);
     assert_eq!(nil_block_child.round(), 2);
@@ -93,6 +94,7 @@ fn test_block_relation() {
         aptos_infallible::duration_since_epoch().as_micros() as u64,
         quorum_cert,
         &signer,
+        Vec::new(),
     );
     assert_eq!(next_block.round(), 1);
     assert_eq!(genesis_block.is_parent_of(&next_block), true);
@@ -121,6 +123,7 @@ fn test_same_qc_different_authors() {
         current_timestamp,
         genesis_qc.clone(),
         &signer,
+        Vec::new(),
     );
 
     let signature = signer.sign(genesis_qc.ledger_info().ledger_info());
@@ -134,10 +137,17 @@ fn test_same_qc_different_authors() {
         current_timestamp,
         genesis_qc_altered,
         &signer,
+        Vec::new(),
     );
 
-    let block_round_1_same =
-        Block::new_proposal(payload, round, current_timestamp, genesis_qc, &signer);
+    let block_round_1_same = Block::new_proposal(
+        payload,
+        round,
+        current_timestamp,
+        genesis_qc,
+        &signer,
+        Vec::new(),
+    );
 
     assert!(block_round_1.id() != block_round_1_altered.id());
     assert_eq!(block_round_1.id(), block_round_1_same.id());
@@ -166,6 +176,7 @@ fn test_block_metadata_bitmaps() {
         start_timestamp,
         genesis_qc,
         &signers[0],
+        Vec::new(),
     );
     let block_metadata_1 = block_1.new_block_metadata(&validators);
     assert_eq!(signers[0].author(), block_metadata_1.proposer());
@@ -201,6 +212,7 @@ fn test_block_metadata_bitmaps() {
         start_timestamp + 1,
         qc_1,
         &signers[1],
+        Vec::new(),
     );
     let block_metadata_2 = block_2.new_block_metadata(&validators);
     assert_eq!(signers[1].author(), block_metadata_2.proposer());
@@ -215,3 +227,69 @@ fn test_nil_block_metadata_bitmaps() {
     assert_eq!(AccountAddress::ZERO, nil_block_metadata.proposer());
     assert_eq!(0, nil_block_metadata.previous_block_votes().len());
 }
+
+#[test]
+fn test_failed_authors_well_formed() {
+    let signer = ValidatorSigner::random(None);
+    let other = Author::random();
+    // Test genesis and the next block
+    let quorum_cert = certificate_for_genesis();
+    let payload = Payload::new_empty();
+
+    let create_block = |round: Round, failed_authors: Vec<(Round, Author)>| {
+        Block::new_proposal(
+            payload.clone(),
+            round,
+            1,
+            quorum_cert.clone(),
+            &signer,
+            failed_authors,
+        )
+    };
+
+    assert!(create_block(1, vec![]).verify_well_formed().is_ok());
+    assert!(create_block(2, vec![]).verify_well_formed().is_ok());
+    assert!(create_block(2, vec![(1, other)])
+        .verify_well_formed()
+        .is_ok());
+    assert!(create_block(3, vec![(1, other)])
+        .verify_well_formed()
+        .is_ok());
+    assert!(create_block(3, vec![(2, other)])
+        .verify_well_formed()
+        .is_ok());
+    assert!(create_block(3, vec![(1, other), (2, other)])
+        .verify_well_formed()
+        .is_ok());
+
+    assert!(create_block(1, vec![(0, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(2, vec![(0, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(2, vec![(2, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(2, vec![(1, other), (1, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(0, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(3, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(4, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(1, other), (1, other), (1, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(1, other), (1, other)])
+        .verify_well_formed()
+        .is_err());
+    assert!(create_block(3, vec![(2, other), (1, other)])
+        .verify_well_formed()
+        .is_err());
+}

consensus/consensus-types/src/block_test_utils.rs

@@ -46,6 +46,7 @@ prop_compose! {
             aptos_infallible::duration_since_epoch().as_micros() as u64,
             parent_qc,
             &signer,
+            Vec::new(),
         )
     }
 }
@@ -87,6 +88,7 @@ prop_compose! {
                 block_data: BlockData::new_proposal(
                     block.payload().unwrap().clone(),
                     block.author().unwrap(),
+                    (*block.block_data().failed_authors().unwrap()).clone(),
                     block.round(),
                     aptos_infallible::duration_since_epoch().as_micros() as u64,
                     block.quorum_cert().clone(),

consensus/safety-rules/src/fuzzing_utils.rs

@@ -98,7 +98,8 @@ prop_compose! {
     ) -> BlockType {
         BlockType::Proposal{
             payload: Payload::DirectMempool(txns),
-            author
+            author,
+            failed_authors: Vec::new(),
         }
     }
 }

consensus/safety-rules/src/test_utils.rs

@@ -68,6 +68,7 @@ pub fn make_proposal_with_qc_and_proof(
             qc.certified_block().timestamp_usecs() + 1,
             qc,
             validator_signer,
+            Vec::new(),
         ),
         None,
         false,

consensus/src/block_storage/block_store_test.rs

@@ -350,6 +350,7 @@ async fn test_illegal_timestamp() {
         genesis.timestamp_usecs(),
         certificate_for_genesis(),
         &signer,
+        Vec::new(),
     );
     let result = block_store
         .execute_and_insert_block(block_with_illegal_timestamp)

consensus/src/epoch_manager.rs

@@ -69,8 +69,8 @@ use std::{
 };
 
 /// Range of rounds (window) that we might be calling proposer election
-/// functions with at any given time.
-const PROPSER_ELECTION_CACHING_WINDOW: usize = 3;
+/// functions with at any given time, in addition to the proposer history length.
+const PROPSER_ELECTION_CACHING_WINDOW_ADDITION: usize = 3;
 
 #[allow(clippy::large_enum_variant)]
 pub enum LivenessStorageData {
@@ -218,7 +218,8 @@ impl EpochManager {
                 // LeaderReputation is not cheap, so we can cache the amount of rounds round_manager needs.
                 Box::new(CachedProposerElection::new(
                     proposer_election,
-                    PROPSER_ELECTION_CACHING_WINDOW,
+                    self.config.max_failed_authors_to_store
+                        + PROPSER_ELECTION_CACHING_WINDOW_ADDITION,
                 ))
             }
             ConsensusProposerType::RoundProposer(round_proposers) => {
@@ -490,6 +491,7 @@ impl EpochManager {
             Arc::new(payload_manager),
             self.time_service.clone(),
             self.config.max_block_size,
+            self.config.max_failed_authors_to_store,
         );
 
         let mut round_manager = RoundManager::new(

consensus/src/experimental/tests/execution_phase_tests.rs

@@ -36,7 +36,14 @@ fn add_execution_phase_test_cases(
 ) {
     let genesis_qc = certificate_for_genesis();
     let (signers, _validators) = random_validator_verifier(1, None, false);
-    let block = Block::new_proposal(Payload::new_empty(), 1, 1, genesis_qc, &signers[0]);
+    let block = Block::new_proposal(
+        Payload::new_empty(),
+        1,
+        1,
+        genesis_qc,
+        &signers[0],
+        Vec::new(),
+    );
 
     // happy path
     phase_tester.add_test_case(
@@ -64,7 +71,8 @@ fn add_execution_phase_test_cases(
         &LedgerInfo::mock_genesis(None),
         random_hash_value,
     );
-    let bad_block = Block::new_proposal(Payload::new_empty(), 1, 1, bad_qc, &signers[0]);
+    let bad_block =
+        Block::new_proposal(Payload::new_empty(), 1, 1, bad_qc, &signers[0], Vec::new());
     phase_tester.add_test_case(
         ExecutionRequest {
             ordered_blocks: vec![ExecutedBlock::new(