[fastpath] support locking transactions and returning effects without…

… signatures (#20128)

## Description 

- Allow locking transactions without getting back a signed transaction.
- Allow getting back unsigned effects for previously executed
transactions.
- Add basic types to support using finalized effects without aggregated
effects sigs.
- Handle `ConsensusTransactionKind::UserTransaction` in callsites that
already handle `ConsensusTransactionKind::CertifiedTransaction`.
- Support reconfiguring components other than `QuorumDriver`, in
reconfig observers.
- Remove some unused code.

## Test plan 

CI

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

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,