Consume system transactions in StateComputer (aptos-labs#11082)

* schedule_compute includes sys txns

* commit should discover system txns

* lint

* misc fix

* misc fix

* fmt

consensus/consensus-types/src/block.rs

@@ -411,26 +411,25 @@ impl Block {
     pub fn transactions_to_execute(
         &self,
         validators: &[AccountAddress],
+        sys_txns: Vec<SystemTransaction>,
         txns: Vec<SignedTransaction>,
         is_block_gas_limit: bool,
     ) -> Vec<Transaction> {
+        let txns = once(Transaction::BlockMetadata(
+            self.new_block_metadata(validators),
+        ))
+        .chain(sys_txns.into_iter().map(Transaction::SystemTransaction))
+        .chain(txns.into_iter().map(Transaction::UserTransaction));
+
         if is_block_gas_limit {
             // After the per-block gas limit change, StateCheckpoint txn
             // is inserted after block execution
-            once(Transaction::BlockMetadata(
-                self.new_block_metadata(validators),
-            ))
-            .chain(txns.into_iter().map(Transaction::UserTransaction))
-            .collect()
+            txns.collect()
         } else {
             // Before the per-block gas limit change, StateCheckpoint txn
             // is inserted here for compatibility.
-            once(Transaction::BlockMetadata(
-                self.new_block_metadata(validators),
-            ))
-            .chain(txns.into_iter().map(Transaction::UserTransaction))
-            .chain(once(Transaction::StateCheckpoint(self.id)))
-            .collect()
+            txns.chain(once(Transaction::StateCheckpoint(self.id)))
+                .collect()
         }
     }
 

consensus/consensus-types/src/block_data.rs

@@ -232,6 +232,19 @@ impl BlockData {
         }
     }
 
+    #[cfg(any(test, feature = "fuzzing"))]
+    pub fn dummy_with_system_txns(sys_txns: Vec<SystemTransaction>) -> Self {
+        Self::new_proposal_ext(
+            sys_txns,
+            Payload::empty(false),
+            Author::ONE,
+            vec![],
+            1,
+            1,
+            QuorumCert::dummy(),
+        )
+    }
+
     pub fn new_genesis(timestamp_usecs: u64, quorum_cert: QuorumCert) -> Self {
         assume!(quorum_cert.certified_block().epoch() < u64::max_value()); // unlikely to be false in this universe
         Self {

consensus/consensus-types/src/executed_block.rs

@@ -119,6 +119,7 @@ impl ExecutedBlock {
     pub fn transactions_to_commit(
         &self,
         validators: &[AccountAddress],
+        sys_txns: Vec<SystemTransaction>,
         txns: Vec<SignedTransaction>,
         is_block_gas_limit: bool,
     ) -> Vec<Transaction> {
@@ -130,7 +131,7 @@ impl ExecutedBlock {
 
         let mut txns_with_state_checkpoint =
             self.block
-                .transactions_to_execute(validators, txns, is_block_gas_limit);
+                .transactions_to_execute(validators, sys_txns, txns, is_block_gas_limit);
         if is_block_gas_limit && !self.state_compute_result.has_reconfiguration() {
             // After the per-block gas limit change,
             // insert state checkpoint at the position

consensus/consensus-types/src/quorum_cert.rs

@@ -41,6 +41,17 @@ impl QuorumCert {
         }
     }
 
+    #[cfg(any(test, feature = "fuzzing"))]
+    pub fn dummy() -> Self {
+        Self {
+            vote_data: VoteData::dummy(),
+            signed_ledger_info: LedgerInfoWithSignatures::new(
+                LedgerInfo::dummy(),
+                AggregateSignature::empty(),
+            ),
+        }
+    }
+
     pub fn vote_data(&self) -> &VoteData {
         &self.vote_data
     }

consensus/consensus-types/src/vote_data.rs

@@ -39,6 +39,14 @@ impl VoteData {
         Self { proposed, parent }
     }
 
+    #[cfg(any(test, feature = "fuzzing"))]
+    pub fn dummy() -> Self {
+        Self {
+            proposed: BlockInfo::empty(),
+            parent: BlockInfo::empty(),
+        }
+    }
+
     /// Returns block information associated to the block being extended by the proposal.
     pub fn parent(&self) -> &BlockInfo {
         &self.parent

consensus/src/lib.rs

@@ -36,6 +36,8 @@ pub mod quorum_store;
 mod recovery_manager;
 mod round_manager;
 mod state_computer;
+#[cfg(test)]
+mod state_computer_tests;
 mod state_replication;
 #[cfg(any(test, feature = "fuzzing"))]
 mod test_utils;

consensus/src/state_computer.rs

@@ -133,14 +133,15 @@ impl StateComputer for ExecutionProxy {
         let txn_deduper = self.transaction_deduper.lock().as_ref().unwrap().clone();
         let txn_shuffler = self.transaction_shuffler.lock().as_ref().unwrap().clone();
         let txn_notifier = self.txn_notifier.clone();
-        let txns = match payload_manager.get_transactions(block).await {
+        let sys_txns = block.sys_txns().cloned().unwrap_or_default();
+        let user_txns = match payload_manager.get_transactions(block).await {
             Ok(txns) => txns,
             Err(err) => return Box::pin(async move { Err(err) }),
         };
 
-        let filtered_txns = self
-            .transaction_filter
-            .filter(block_id, block.timestamp_usecs(), txns);
+        let filtered_txns =
+            self.transaction_filter
+                .filter(block_id, block.timestamp_usecs(), user_txns);
         let deduped_txns = txn_deduper.dedup(filtered_txns);
         let shuffled_txns = txn_shuffler.shuffle(deduped_txns);
 
@@ -150,6 +151,7 @@ impl StateComputer for ExecutionProxy {
         let timestamp = block.timestamp_usecs();
         let transactions_to_execute = block.transactions_to_execute(
             &self.validators.lock(),
+            sys_txns,
             shuffled_txns.clone(),
             block_executor_onchain_config.has_any_block_gas_limit(),
         );
@@ -222,6 +224,7 @@ impl StateComputer for ExecutionProxy {
                 payloads.push(payload.clone());
             }
 
+            let sys_txns = block.sys_txns().map_or(vec![], Vec::clone);
             let signed_txns = payload_manager.get_transactions(block.block()).await?;
             let filtered_txns =
                 self.transaction_filter
@@ -231,6 +234,7 @@ impl StateComputer for ExecutionProxy {
 
             txns.extend(block.transactions_to_commit(
                 &self.validators.lock(),
+                sys_txns,
                 shuffled_txns,
                 block_executor_onchain_config.has_any_block_gas_limit(),
             ));