make SubmitToConsensus trait non-async (#21016)

---

## 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): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

crates/sui-core/src/authority/execution_time_estimator.rs

@@ -162,7 +162,6 @@ impl ExecutionTimeObserver {
                 if let Err(e) = self
                     .consensus_adapter
                     .submit_to_consensus(&[transaction], &epoch_store)
-                    .await
                 {
                     if !matches!(e, SuiError::EpochEnded(_)) {
                         warn!("failed to submit execution time observation: {e:?}");

crates/sui-core/src/checkpoints/checkpoint_output.rs

@@ -110,8 +110,7 @@ impl<T: SubmitToConsensus + ReconfigurationInitiator> CheckpointOutput
             let message = CheckpointSignatureMessage { summary };
             let transaction = ConsensusTransaction::new_checkpoint_signature_message(message);
             self.sender
-                .submit_to_consensus(&vec![transaction], epoch_store)
-                .await?;
+                .submit_to_consensus(&vec![transaction], epoch_store)?;
             self.metrics
                 .last_sent_checkpoint_signature
                 .set(checkpoint_seq as i64);

crates/sui-core/src/consensus_adapter.rs

@@ -203,9 +203,8 @@ pub trait ConsensusOverloadChecker: Sync + Send + 'static {
 pub type BlockStatusReceiver = oneshot::Receiver<BlockStatus>;
 
 #[mockall::automock]
-#[async_trait::async_trait]
 pub trait SubmitToConsensus: Sync + Send + 'static {
-    async fn submit_to_consensus(
+    fn submit_to_consensus(
         &self,
         transactions: &[ConsensusTransaction],
         epoch_store: &Arc<AuthorityPerEpochStore>,
@@ -1273,9 +1272,8 @@ impl<'a> Drop for InflightDropGuard<'a> {
     }
 }
 
-#[async_trait::async_trait]
 impl SubmitToConsensus for Arc<ConsensusAdapter> {
-    async fn submit_to_consensus(
+    fn submit_to_consensus(
         &self,
         transactions: &[ConsensusTransaction],
         epoch_store: &Arc<AuthorityPerEpochStore>,

crates/sui-core/src/epoch/randomness.rs

@@ -424,8 +424,7 @@ impl RandomnessManager {
         });
         if !fail_point_skip_sending {
             self.consensus_adapter
-                .submit_to_consensus(&[transaction], &epoch_store)
-                .await?;
+                .submit_to_consensus(&[transaction], &epoch_store)?;
         }
 
         epoch_store
@@ -495,8 +494,7 @@ impl RandomnessManager {
                     });
                     if !fail_point_skip_sending {
                         self.consensus_adapter
-                            .submit_to_consensus(&[transaction], &epoch_store)
-                            .await?;
+                            .submit_to_consensus(&[transaction], &epoch_store)?;
                     }
 
                     let elapsed = self.dkg_start_time.get().map(|t| t.elapsed().as_millis());

crates/sui-core/src/mock_consensus.rs

@@ -9,7 +9,7 @@ use crate::consensus_handler::SequencedConsensusTransaction;
 use consensus_core::BlockRef;
 use prometheus::Registry;
 use std::sync::{Arc, Weak};
-use sui_types::error::{SuiError, SuiResult};
+use sui_types::error::SuiResult;
 use sui_types::executable_transaction::VerifiedExecutableTransaction;
 use sui_types::messages_consensus::{ConsensusTransaction, ConsensusTransactionKind};
 use sui_types::transaction::{VerifiedCertificate, VerifiedTransaction};
@@ -96,18 +96,27 @@ impl MockConsensusClient {
             }
         }
     }
+
+    fn submit_impl(&self, transactions: &[ConsensusTransaction]) -> SuiResult<BlockStatusReceiver> {
+        // TODO: maybe support multi-transactions and remove this check
+        assert!(transactions.len() == 1);
+        let transaction = &transactions[0];
+        self.tx_sender
+            .try_send(transaction.clone())
+            .expect("MockConsensusClient channel should not overflow");
+        Ok(with_block_status(consensus_core::BlockStatus::Sequenced(
+            BlockRef::MIN,
+        )))
+    }
 }
 
-#[async_trait::async_trait]
 impl SubmitToConsensus for MockConsensusClient {
-    async fn submit_to_consensus(
+    fn submit_to_consensus(
         &self,
         transactions: &[ConsensusTransaction],
-        epoch_store: &Arc<AuthorityPerEpochStore>,
+        _epoch_store: &Arc<AuthorityPerEpochStore>,
     ) -> SuiResult {
-        self.submit(transactions, epoch_store)
-            .await
-            .map(|_response| ())
+        self.submit_impl(transactions).map(|_response| ())
     }
 }
 
@@ -118,16 +127,7 @@ impl ConsensusClient for MockConsensusClient {
         transactions: &[ConsensusTransaction],
         _epoch_store: &Arc<AuthorityPerEpochStore>,
     ) -> SuiResult<BlockStatusReceiver> {
-        // TODO: maybe support multi-transactions and remove this check
-        assert!(transactions.len() == 1);
-        let transaction = &transactions[0];
-        self.tx_sender
-            .send(transaction.clone())
-            .await
-            .map_err(|e| SuiError::Unknown(e.to_string()))?;
-        Ok(with_block_status(consensus_core::BlockStatus::Sequenced(
-            BlockRef::MIN,
-        )))
+        self.submit_impl(transactions)
     }
 }