[fastpath] support locking transactions and returning effects without signatures

consensus/core/src/block_verifier.rs

@@ -264,6 +264,7 @@ mod test {
 
     struct TxnSizeVerifier {}
 
+    #[async_trait::async_trait]
     impl TransactionVerifier for TxnSizeVerifier {
         // Fails verification if any transaction is < 4 bytes.
         fn verify_batch(&self, transactions: &[&[u8]]) -> Result<(), ValidationError> {
@@ -278,8 +279,11 @@ mod test {
             Ok(())
         }
 
-        fn vote_batch(&self, _batch: &[&[u8]]) -> Vec<TransactionIndex> {
-            vec![]
+        async fn verify_and_vote_batch(
+            &self,
+            _batch: &[&[u8]],
+        ) -> Result<Vec<TransactionIndex>, ValidationError> {
+            Ok(vec![])
         }
     }
 

consensus/core/src/transaction.rs

@@ -210,16 +210,20 @@ impl TransactionClient {
 
 /// `TransactionVerifier` implementation is supplied by Sui to validate transactions in a block,
 /// before acceptance of the block.
+#[async_trait::async_trait]
 pub trait TransactionVerifier: Send + Sync + 'static {
     /// Determines if this batch of transactions is valid.
     /// Fails if any one of the transactions is invalid.
     fn verify_batch(&self, batch: &[&[u8]]) -> Result<(), ValidationError>;
 
-    /// Returns indices of transactions to reject.
-    /// Currently only uncertified user transactions can be rejected.
-    /// The rest of transactions are implicitly voted to accept.
-    // TODO: add rejection reasons, add VoteError and wrap the return in Result<>.
-    fn vote_batch(&self, batch: &[&[u8]]) -> Vec<TransactionIndex>;
+    /// Returns indices of transactions to reject, validator error over transactions.
+    /// Currently only uncertified user transactions can be rejected. The rest of transactions
+    /// are implicitly voted to be accepted.
+    /// When the result is an error, the whole block should be rejected from local DAG instead.
+    async fn verify_and_vote_batch(
+        &self,
+        batch: &[&[u8]],
+    ) -> Result<Vec<TransactionIndex>, ValidationError>;
 }
 
 #[derive(Debug, Error)]
@@ -229,16 +233,21 @@ pub enum ValidationError {
 }
 
 /// `NoopTransactionVerifier` accepts all transactions.
-#[allow(unused)]
+#[cfg(test)]
 pub(crate) struct NoopTransactionVerifier;
 
+#[cfg(test)]
+#[async_trait::async_trait]
 impl TransactionVerifier for NoopTransactionVerifier {
     fn verify_batch(&self, _batch: &[&[u8]]) -> Result<(), ValidationError> {
         Ok(())
     }
 
-    fn vote_batch(&self, _batch: &[&[u8]]) -> Vec<TransactionIndex> {
-        vec![]
+    async fn verify_and_vote_batch(
+        &self,
+        _batch: &[&[u8]],
+    ) -> Result<Vec<TransactionIndex>, ValidationError> {
+        Ok(vec![])
     }
 }
 

crates/sui-benchmark/src/embedded_reconfig_observer.rs

@@ -5,9 +5,9 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use std::sync::Arc;
 use sui_core::authority_aggregator::AuthorityAggregator;
+use sui_core::quorum_driver::AuthorityAggregatorUpdatable;
 use sui_core::{
-    authority_client::NetworkAuthorityClient,
-    quorum_driver::{reconfig_observer::ReconfigObserver, QuorumDriver},
+    authority_client::NetworkAuthorityClient, quorum_driver::reconfig_observer::ReconfigObserver,
 };
 use sui_network::default_mysten_network_config;
 use sui_types::sui_system_state::SuiSystemStateTrait;
@@ -79,12 +79,15 @@ impl ReconfigObserver<NetworkAuthorityClient> for EmbeddedReconfigObserver {
         Box::new(self.clone())
     }
 
-    async fn run(&mut self, quorum_driver: Arc<QuorumDriver<NetworkAuthorityClient>>) {
+    async fn run(
+        &mut self,
+        updatable: Arc<dyn AuthorityAggregatorUpdatable<NetworkAuthorityClient>>,
+    ) {
         loop {
             tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
-            let auth_agg = quorum_driver.authority_aggregator().load();
+            let auth_agg = updatable.authority_aggregator();
             match self.get_committee(auth_agg.clone()).await {
-                Ok(new_auth_agg) => quorum_driver.update_validators(new_auth_agg).await,
+                Ok(new_auth_agg) => updatable.update_authority_aggregator(new_auth_agg),
                 Err(err) => {
                     error!(
                         "Failed to recreate authority aggregator with committee: {}",

crates/sui-benchmark/src/fullnode_reconfig_observer.rs

@@ -7,7 +7,7 @@ use sui_core::{
     authority_aggregator::{AuthAggMetrics, AuthorityAggregator},
     authority_client::NetworkAuthorityClient,
     epoch::committee_store::CommitteeStore,
-    quorum_driver::{reconfig_observer::ReconfigObserver, QuorumDriver},
+    quorum_driver::{reconfig_observer::ReconfigObserver, AuthorityAggregatorUpdatable},
     safe_client::SafeClientMetricsBase,
 };
 use sui_sdk::{SuiClient, SuiClientBuilder};
@@ -56,7 +56,7 @@ impl ReconfigObserver<NetworkAuthorityClient> for FullNodeReconfigObserver {
         Box::new(self.clone())
     }
 
-    async fn run(&mut self, quorum_driver: Arc<QuorumDriver<NetworkAuthorityClient>>) {
+    async fn run(&mut self, driver: Arc<dyn AuthorityAggregatorUpdatable<NetworkAuthorityClient>>) {
         loop {
             tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
             match self
@@ -67,7 +67,7 @@ impl ReconfigObserver<NetworkAuthorityClient> for FullNodeReconfigObserver {
             {
                 Ok(sui_system_state) => {
                     let epoch_id = sui_system_state.epoch;
-                    if epoch_id > quorum_driver.current_epoch() {
+                    if epoch_id > driver.epoch() {
                         debug!(epoch_id, "Got SuiSystemState in newer epoch");
                         let new_committee = sui_system_state.get_sui_committee_for_benchmarking();
                         let _ = self
@@ -80,7 +80,7 @@ impl ReconfigObserver<NetworkAuthorityClient> for FullNodeReconfigObserver {
                             self.auth_agg_metrics.clone(),
                             Arc::new(HashMap::new()),
                         );
-                        quorum_driver.update_validators(Arc::new(auth_agg)).await
+                        driver.update_authority_aggregator(Arc::new(auth_agg));
                     } else {
                         trace!(
                             epoch_id,