[Consensus Observer] Add block payload verification.

consensus/consensus-types/src/common.rs

@@ -165,6 +165,10 @@ impl ProofWithData {
         }
     }
 
+    pub fn empty() -> Self {
+        Self::new(vec![])
+    }
+
     pub fn extend(&mut self, other: ProofWithData) {
         let other_data_status = other.status.lock().as_mut().unwrap().take();
         self.proofs.extend(other.proofs);

consensus/src/consensus_observer/missing_blocks.rs

@@ -159,6 +159,7 @@ impl MissingBlockStore {
 #[cfg(test)]
 mod test {
     use super::*;
+    use crate::consensus_observer::network_message::BlockTransactionPayload;
     use aptos_consensus_types::{
         block::Block,
         block_data::{BlockData, BlockType},
@@ -378,7 +379,7 @@ mod test {
         );
 
         // Create a new block payload store and insert payloads for the second block
-        let mut block_payload_store = BlockPayloadStore::new();
+        let mut block_payload_store = BlockPayloadStore::new(consensus_observer_config);
         let second_block = missing_blocks[1].clone();
         insert_payloads_for_ordered_block(&mut block_payload_store, &second_block);
 
@@ -439,13 +440,17 @@ mod test {
         );
 
         // Create an empty block payload store
-        let mut block_payload_store = BlockPayloadStore::new();
+        let mut block_payload_store = BlockPayloadStore::new(consensus_observer_config);
 
         // Incrementally insert and process each payload for the first block
         let first_block = missing_blocks.first().unwrap().clone();
         for block in first_block.blocks().clone() {
             // Insert the block
-            block_payload_store.insert_block_payload(block.block_info(), vec![], None);
+            block_payload_store.insert_block_payload(
+                block.block_info(),
+                BlockTransactionPayload::empty(),
+                true,
+            );
 
             // Attempt to remove the block (which might not be ready)
             let payload_round = block.round();
@@ -486,7 +491,11 @@ mod test {
             // Insert the block only if this is not the first block
             let payload_round = block.round();
             if payload_round != last_block.first_block().round() {
-                block_payload_store.insert_block_payload(block.block_info(), vec![], None);
+                block_payload_store.insert_block_payload(
+                    block.block_info(),
+                    BlockTransactionPayload::empty(),
+                    true,
+                );
             }
 
             // Attempt to remove the block (which might not be ready)
@@ -533,7 +542,7 @@ mod test {
         );
 
         // Create a new block payload store and insert payloads for the first block
-        let mut block_payload_store = BlockPayloadStore::new();
+        let mut block_payload_store = BlockPayloadStore::new(consensus_observer_config);
         let first_block = missing_blocks.first().unwrap().clone();
         insert_payloads_for_ordered_block(&mut block_payload_store, &first_block);
 
@@ -614,7 +623,7 @@ mod test {
         );
 
         // Create an empty block payload store
-        let block_payload_store = BlockPayloadStore::new();
+        let block_payload_store = BlockPayloadStore::new(consensus_observer_config);
 
         // Remove the third block (which is not ready)
         let third_block = missing_blocks[2].clone();
@@ -716,7 +725,11 @@ mod test {
         ordered_block: &OrderedBlock,
     ) {
         for block in ordered_block.blocks() {
-            block_payload_store.insert_block_payload(block.block_info(), vec![], None);
+            block_payload_store.insert_block_payload(
+                block.block_info(),
+                BlockTransactionPayload::empty(),
+                true,
+            );
         }
     }
 

consensus/src/consensus_observer/network_message.rs

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::consensus_observer::error::Error;
-use aptos_consensus_types::pipelined_block::PipelinedBlock;
+use aptos_consensus_types::{
+    common::ProofWithData, pipelined_block::PipelinedBlock, proof_of_store::BatchInfo,
+};
 use aptos_types::{
     block_info::{BlockInfo, Round},
     epoch_change::Verifier,
@@ -46,13 +48,11 @@ impl ConsensusObserverMessage {
     /// Creates and returns a new block payload message using the given block, transactions and limit
     pub fn new_block_payload_message(
         block: BlockInfo,
-        transactions: Vec<SignedTransaction>,
-        limit: Option<u64>,
+        transaction_payload: BlockTransactionPayload,
     ) -> ConsensusObserverDirectSend {
         ConsensusObserverDirectSend::BlockPayload(BlockPayload {
             block,
-            transactions,
-            limit,
+            transaction_payload,
         })
     }
 }
@@ -150,10 +150,11 @@ impl ConsensusObserverDirectSend {
             },
             ConsensusObserverDirectSend::BlockPayload(block_payload) => {
                 format!(
-                    "BlockPayload: {} {} {:?}",
-                    block_payload.block.id(),
-                    block_payload.transactions.len(),
-                    block_payload.limit
+                    "BlockPayload: {}. Number of transactions: {}, limit: {:?}, proofs: {:?}",
+                    block_payload.block,
+                    block_payload.transaction_payload.transactions.len(),
+                    block_payload.transaction_payload.limit,
+                    block_payload.transaction_payload.proof_with_data.proofs,
                 )
             },
         }
@@ -304,10 +305,45 @@ impl CommitDecision {
     }
 }
 
-/// Payload message contains the block, transactions and the limit of the block
+/// The transaction payload of each block
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-pub struct BlockPayload {
-    pub block: BlockInfo,
+pub struct BlockTransactionPayload {
     pub transactions: Vec<SignedTransaction>,
     pub limit: Option<u64>,
+    pub proof_with_data: ProofWithData,
+    pub inline_batches: Vec<BatchInfo>,
+}
+
+impl BlockTransactionPayload {
+    pub fn new(
+        transactions: Vec<SignedTransaction>,
+        limit: Option<u64>,
+        proof_with_data: ProofWithData,
+        inline_batches: Vec<BatchInfo>,
+    ) -> Self {
+        Self {
+            transactions,
+            limit,
+            proof_with_data,
+            inline_batches,
+        }
+    }
+
+    #[cfg(test)]
+    /// Returns an empty transaction payload (for testing)
+    pub fn empty() -> Self {
+        Self {
+            transactions: vec![],
+            limit: None,
+            proof_with_data: ProofWithData::empty(),
+            inline_batches: vec![],
+        }
+    }
+}
+
+/// Payload message contains the block and transaction payload
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct BlockPayload {
+    pub block: BlockInfo,
+    pub transaction_payload: BlockTransactionPayload,
 }