diff --git a/crates/sui-core/src/authority.rs b/crates/sui-core/src/authority.rs
index add2ec24013ff..2156a0ba077c3 100644
--- a/crates/sui-core/src/authority.rs
+++ b/crates/sui-core/src/authority.rs
@@ -228,9 +228,11 @@ pub struct AuthorityMetrics {
     batch_size: Histogram,
 
     authority_state_handle_transaction_latency: Histogram,
+    authority_state_handle_transaction_v2_latency: Histogram,
 
     execute_certificate_latency_single_writer: Histogram,
     execute_certificate_latency_shared_object: Histogram,
+    await_transaction_latency: Histogram,
 
     execute_certificate_with_effects_latency: Histogram,
     internal_execution_latency: Histogram,
@@ -434,8 +436,22 @@ impl AuthorityMetrics {
                 registry,
             )
             .unwrap(),
+            authority_state_handle_transaction_v2_latency: register_histogram_with_registry!(
+                "authority_state_handle_transaction_v2_latency",
+                "Latency of handling transactions with v2",
+                LATENCY_SEC_BUCKETS.to_vec(),
+                registry,
+            )
+            .unwrap(),
             execute_certificate_latency_single_writer,
             execute_certificate_latency_shared_object,
+            await_transaction_latency: register_histogram_with_registry!(
+                "await_transaction_latency",
+                "Latency of awaiting user transaction execution, including waiting for inputs",
+                LATENCY_SEC_BUCKETS.to_vec(),
+                registry,
+            )
+            .unwrap(),
             execute_certificate_with_effects_latency: register_histogram_with_registry!(
                 "authority_state_execute_certificate_with_effects_latency",
                 "Latency of executing certificates with effects, including waiting for inputs",
@@ -839,17 +855,11 @@ impl AuthorityState {
         self.checkpoint_store.get_epoch_state_commitments(epoch)
     }
 
-    /// This is a private method and should be kept that way. It doesn't check whether
-    /// the provided transaction is a system transaction, and hence can only be called internally.
-    #[instrument(level = "trace", skip_all)]
-    async fn handle_transaction_impl(
+    fn handle_transaction_deny_checks(
         &self,
-        transaction: VerifiedTransaction,
+        transaction: &VerifiedTransaction,
         epoch_store: &Arc<AuthorityPerEpochStore>,
-    ) -> SuiResult<VerifiedSignedTransaction> {
-        // Ensure that validator cannot reconfigure while we are signing the tx
-        let _execution_lock = self.execution_lock_for_signing().await;
-
+    ) -> SuiResult<CheckedInputObjects> {
         let tx_digest = transaction.digest();
         let tx_data = transaction.data().transaction_data();
 
@@ -903,6 +913,23 @@ impl AuthorityState {
             )?;
         }
 
+        Ok(checked_input_objects)
+    }
+
+    /// This is a private method and should be kept that way. It doesn't check whether
+    /// the provided transaction is a system transaction, and hence can only be called internally.
+    #[instrument(level = "trace", skip_all)]
+    async fn handle_transaction_impl(
+        &self,
+        transaction: VerifiedTransaction,
+        epoch_store: &Arc<AuthorityPerEpochStore>,
+    ) -> SuiResult<VerifiedSignedTransaction> {
+        // Ensure that validator cannot reconfigure while we are signing the tx
+        let _execution_lock = self.execution_lock_for_signing().await;
+
+        let checked_input_objects =
+            self.handle_transaction_deny_checks(&transaction, epoch_store)?;
+
         let owned_objects = checked_input_objects.inner().filter_owned_objects();
 
         let signed_transaction = VerifiedSignedTransaction::new(
@@ -974,6 +1001,51 @@ impl AuthorityState {
         }
     }
 
+    #[instrument(level = "trace", skip_all)]
+    pub async fn handle_transaction_v2(
+        &self,
+        epoch_store: &Arc<AuthorityPerEpochStore>,
+        transaction: VerifiedTransaction,
+    ) -> SuiResult<Option<(SenderSignedData, TransactionStatus)>> {
+        let tx_digest = *transaction.digest();
+        debug!("handle_transaction_v2");
+
+        // Ensure an idempotent answer.
+        let tx_status = self.get_transaction_status(&tx_digest, epoch_store)?;
+        if tx_status.is_some() {
+            return Ok(tx_status);
+        }
+
+        let _metrics_guard = self
+            .metrics
+            .authority_state_handle_transaction_v2_latency
+            .start_timer();
+        self.metrics.tx_orders.inc();
+
+        // The should_accept_user_certs check here is best effort, because
+        // between a validator signs a tx and a cert is formed, the validator
+        // could close the window.
+        if !epoch_store
+            .get_reconfig_state_read_lock_guard()
+            .should_accept_user_certs()
+        {
+            return Err(SuiError::ValidatorHaltedAtEpochEnd);
+        }
+
+        match self.handle_transaction_impl(transaction, epoch_store).await {
+            // TODO(fastpath): We don't actually need the signed transaction here but just call
+            // into this function to acquire locks. Consider refactoring to avoid the extra work.
+            Ok(_signed) => Ok(None),
+            // It happens frequently that while we are checking the validity of the transaction, it
+            // has just been executed.
+            // In that case, we could still return Ok to avoid showing confusing errors.
+            Err(e) => self
+                .get_transaction_status(&tx_digest, epoch_store)?
+                .ok_or(e)
+                .map(Some),
+        }
+    }
+
     pub fn check_system_overload_at_signing(&self) -> bool {
         self.config
             .authority_overload_config
@@ -1127,7 +1199,21 @@ impl AuthorityState {
             self.enqueue_certificates_for_execution(vec![certificate.clone()], epoch_store);
         }
 
-        self.notify_read_effects(certificate).await
+        self.notify_read_effects(*certificate.digest()).await
+    }
+
+    /// Awaits the effects of executing a user transaction.
+    ///
+    /// Relies on consensus to enqueue the transaction for execution.
+    pub async fn await_transaction_effects(
+        &self,
+        digest: TransactionDigest,
+    ) -> SuiResult<TransactionEffects> {
+        let _metrics_guard = self.metrics.await_transaction_latency.start_timer();
+        debug!("await_transaction");
+
+        // TODO(fastpath): Add handling for transactions rejected by Mysticeti fast path.
+        self.notify_read_effects(digest).await
     }
 
     /// Internal logic to execute a certificate.
@@ -1219,10 +1305,10 @@ impl AuthorityState {
 
     pub async fn notify_read_effects(
         &self,
-        certificate: &VerifiedCertificate,
+        digest: TransactionDigest,
     ) -> SuiResult<TransactionEffects> {
         self.get_transaction_cache_reader()
-            .notify_read_executed_effects(&[*certificate.digest()])
+            .notify_read_executed_effects(&[digest])
             .await
             .map(|mut r| r.pop().expect("must return correct number of effects"))
     }
diff --git a/crates/sui-core/src/authority/authority_per_epoch_store.rs b/crates/sui-core/src/authority/authority_per_epoch_store.rs
index 8e38a41242e84..967da2b5c54ed 100644
--- a/crates/sui-core/src/authority/authority_per_epoch_store.rs
+++ b/crates/sui-core/src/authority/authority_per_epoch_store.rs
@@ -1965,17 +1965,12 @@ impl AuthorityPerEpochStore {
             .any(|processed| processed))
     }
 
-    /// Check whether any certificates were processed by consensus.
-    /// This handles multiple certificates at once.
-    pub fn is_all_tx_certs_consensus_message_processed<'a>(
+    /// Returns true if all messages with the given keys were processed by consensus.
+    pub fn all_external_consensus_messages_processed(
         &self,
-        certificates: impl Iterator<Item = &'a VerifiedCertificate>,
+        keys: impl Iterator<Item = ConsensusTransactionKey>,
     ) -> SuiResult<bool> {
-        let keys = certificates.map(|cert| {
-            SequencedConsensusTransactionKey::External(ConsensusTransactionKey::Certificate(
-                *cert.digest(),
-            ))
-        });
+        let keys = keys.map(SequencedConsensusTransactionKey::External);
         Ok(self
             .check_consensus_messages_processed(keys)?
             .into_iter()
diff --git a/crates/sui-core/src/authority/authority_test_utils.rs b/crates/sui-core/src/authority/authority_test_utils.rs
index 07d91da570ddd..ab2009c2e82d6 100644
--- a/crates/sui-core/src/authority/authority_test_utils.rs
+++ b/crates/sui-core/src/authority/authority_test_utils.rs
@@ -377,7 +377,7 @@ pub async fn enqueue_all_and_execute_all(
     );
     let mut output = Vec::new();
     for cert in certificates {
-        let effects = authority.notify_read_effects(&cert).await?;
+        let effects = authority.notify_read_effects(*cert.digest()).await?;
         output.push(effects);
     }
     Ok(output)
diff --git a/crates/sui-core/src/authority_server.rs b/crates/sui-core/src/authority_server.rs
index 04e33529ce090..a5ece01381f9b 100644
--- a/crates/sui-core/src/authority_server.rs
+++ b/crates/sui-core/src/authority_server.rs
@@ -19,9 +19,9 @@ use sui_network::{
     api::{Validator, ValidatorServer},
     tonic,
 };
-use sui_types::effects::TransactionEffectsAPI;
-use sui_types::messages_consensus::ConsensusTransaction;
-use sui_types::messages_grpc::{HandleCertificateRequestV3, HandleCertificateResponseV3};
+use sui_types::messages_grpc::{
+    HandleCertificateRequestV3, HandleCertificateResponseV3, HandleTransactionResponseV2,
+};
 use sui_types::messages_grpc::{
     HandleCertificateResponseV2, HandleTransactionResponse, ObjectInfoRequest, ObjectInfoResponse,
     SubmitCertificateResponse, SystemStateRequest, TransactionInfoRequest, TransactionInfoResponse,
@@ -32,6 +32,7 @@ use sui_types::messages_grpc::{
 use sui_types::multiaddr::Multiaddr;
 use sui_types::sui_system_state::SuiSystemState;
 use sui_types::traffic_control::{ClientIdSource, PolicyConfig, RemoteFirewallConfig, Weight};
+use sui_types::{effects::TransactionEffectsAPI, messages_grpc::HandleTransactionRequestV2};
 use sui_types::{error::*, transaction::*};
 use sui_types::{
     fp_ensure,
@@ -39,6 +40,10 @@ use sui_types::{
         CheckpointRequest, CheckpointRequestV2, CheckpointResponse, CheckpointResponseV2,
     },
 };
+use sui_types::{
+    messages_consensus::{ConsensusTransaction, ConsensusTransactionKind},
+    messages_grpc::TransactionStatus,
+};
 use tap::TapFallible;
 use tokio::task::JoinHandle;
 use tonic::metadata::{Ascii, MetadataValue};
@@ -171,10 +176,12 @@ pub struct ValidatorServiceMetrics {
     pub cert_verification_latency: Histogram,
     pub consensus_latency: Histogram,
     pub handle_transaction_latency: Histogram,
+    pub handle_transaction_v2_latency: Histogram,
     pub submit_certificate_consensus_latency: Histogram,
     pub handle_certificate_consensus_latency: Histogram,
     pub handle_certificate_non_consensus_latency: Histogram,
     pub handle_soft_bundle_certificates_consensus_latency: Histogram,
+    pub handle_transaction_consensus_latency: Histogram,
 
     num_rejected_tx_in_epoch_boundary: IntCounter,
     num_rejected_cert_in_epoch_boundary: IntCounter,
@@ -223,6 +230,13 @@ impl ValidatorServiceMetrics {
                 registry,
             )
             .unwrap(),
+            handle_transaction_v2_latency: register_histogram_with_registry!(
+                "validator_service_handle_transaction_v2_latency",
+                "Latency of v2 transaction handler",
+                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
+                registry,
+            )
+            .unwrap(),
             handle_certificate_consensus_latency: register_histogram_with_registry!(
                 "validator_service_handle_certificate_consensus_latency",
                 "Latency of handling a consensus transaction certificate",
@@ -251,6 +265,13 @@ impl ValidatorServiceMetrics {
                 registry,
             )
             .unwrap(),
+            handle_transaction_consensus_latency: register_histogram_with_registry!(
+                "validator_service_handle_transaction_consensus_latency",
+                "Latency of handling a user transaction sent through consensus",
+                mysten_metrics::COARSE_LATENCY_SEC_BUCKETS.to_vec(),
+                registry,
+            )
+            .unwrap(),
             num_rejected_tx_in_epoch_boundary: register_int_counter_with_registry!(
                 "validator_service_num_rejected_tx_in_epoch_boundary",
                 "Number of rejected transaction during epoch transitioning",
@@ -452,6 +473,127 @@ impl ValidatorService {
         Ok((tonic::Response::new(info), Weight::zero()))
     }
 
+    async fn handle_transaction_v2(
+        &self,
+        request: tonic::Request<HandleTransactionRequestV2>,
+    ) -> WrappedServiceResponse<HandleTransactionResponseV2> {
+        let Self {
+            state,
+            consensus_adapter,
+            metrics,
+            traffic_controller: _,
+            client_id_source: _,
+        } = self.clone();
+        let epoch_store = state.load_epoch_store_one_call_per_task();
+        if !epoch_store.protocol_config().mysticeti_fastpath() {
+            return Err(SuiError::UnsupportedFeatureError {
+                error: "Mysticeti fastpath".to_string(),
+            }
+            .into());
+        }
+
+        let HandleTransactionRequestV2 {
+            transaction,
+            include_events,
+            include_input_objects,
+            include_output_objects,
+            include_auxiliary_data,
+        } = request.into_inner();
+
+        transaction.validity_check(epoch_store.protocol_config(), epoch_store.epoch())?;
+
+        // Check system overload
+        let overload_check_res = self.state.check_system_overload(
+            &consensus_adapter,
+            transaction.data(),
+            state.check_system_overload_at_signing(),
+        );
+        if let Err(error) = overload_check_res {
+            metrics
+                .num_rejected_tx_during_overload
+                .with_label_values(&[error.as_ref()])
+                .inc();
+            return Err(error.into());
+        }
+
+        let _handle_tx_metrics_guard = metrics.handle_transaction_v2_latency.start_timer();
+
+        let tx_verif_metrics_guard = metrics.tx_verification_latency.start_timer();
+        let transaction = epoch_store.verify_transaction(transaction).tap_err(|_| {
+            metrics.signature_errors.inc();
+        })?;
+        drop(tx_verif_metrics_guard);
+
+        // Enable Trace Propagation across spans/processes using tx_digest
+        let tx_digest = transaction.digest();
+        let span = error_span!("validator_state_process_tx_v2", ?tx_digest);
+
+        let tx_status = state
+            .handle_transaction_v2(&epoch_store, transaction.clone())
+            .instrument(span)
+            .await
+            .tap_err(|e| {
+                if let SuiError::ValidatorHaltedAtEpochEnd = e {
+                    metrics.num_rejected_tx_in_epoch_boundary.inc();
+                }
+            })?;
+        if let Some((
+            _sender_signed_data,
+            // TODO(fastpath): Suppress duplicate transaction submission in consensus
+            // adapter, if not already done. (If we get back `TransactionStatus::Signed``
+            // here we still need to proceed with submission logic, because the previous
+            // RPC might have been dropped after signing but before submission.)
+            TransactionStatus::Executed(_sign_info, signed_effects, events),
+        )) = tx_status
+        {
+            let input_objects = include_input_objects
+                .then(|| state.get_transaction_input_objects(signed_effects.data()))
+                .and_then(Result::ok);
+            let output_objects = include_output_objects
+                .then(|| state.get_transaction_output_objects(signed_effects.data()))
+                .and_then(Result::ok);
+
+            return Ok((
+                tonic::Response::new(HandleTransactionResponseV2 {
+                    effects: signed_effects.into_data(),
+                    events: include_events.then_some(events),
+                    input_objects,
+                    output_objects,
+                    auxiliary_data: None, // We don't have any aux data generated presently
+                }),
+                Weight::zero(),
+            ));
+        }
+
+        let _latency_metric_guard = metrics.handle_transaction_consensus_latency.start_timer();
+        let span = error_span!("handle_transaction_v2", tx_digest = ?transaction.digest());
+        self.handle_submit_to_consensus(
+            nonempty![ConsensusTransaction::new_user_transaction_message(
+                &self.state.name,
+                transaction.into()
+            )],
+            include_events,
+            include_input_objects,
+            include_output_objects,
+            include_auxiliary_data,
+            &epoch_store,
+            true,
+        )
+        .instrument(span)
+        .await
+        .map(|(resp, spam_weight)| {
+            (
+                tonic::Response::new(
+                    resp.expect(
+                        "handle_submit_to_consensus should not return none with wait_for_effects=true",
+                    )
+                    .remove(0),
+                ),
+                spam_weight,
+            )
+        })
+    }
+
     // In addition to the response from handling the certificates,
     // returns a bool indicating whether the request should be tallied
     // toward spam count. In general, this should be set to true for
@@ -463,7 +605,7 @@ impl ValidatorService {
         include_events: bool,
         include_input_objects: bool,
         include_output_objects: bool,
-        _include_auxiliary_data: bool,
+        include_auxiliary_data: bool,
         epoch_store: &Arc<AuthorityPerEpochStore>,
         wait_for_effects: bool,
     ) -> Result<(Option<Vec<HandleCertificateResponseV3>>, Weight), tonic::Status> {
@@ -556,7 +698,62 @@ impl ValidatorService {
                 .into_iter()
                 .collect::<Result<Vec<_>, _>>()?
         };
+        let consensus_transactions =
+            NonEmpty::collect(verified_certificates.iter().map(|certificate| {
+                ConsensusTransaction::new_certificate_message(
+                    &self.state.name,
+                    certificate.clone().into(),
+                )
+            }))
+            .unwrap();
+
+        let (responses, weight) = self
+            .handle_submit_to_consensus(
+                consensus_transactions,
+                include_events,
+                include_input_objects,
+                include_output_objects,
+                include_auxiliary_data,
+                epoch_store,
+                wait_for_effects,
+            )
+            .await?;
+        // Sign the returned TransactionEffects.
+        let responses = if let Some(responses) = responses {
+            Some(
+                responses
+                    .into_iter()
+                    .map(|response| {
+                        let signed_effects =
+                            self.state.sign_effects(response.effects, epoch_store)?;
+                        Ok(HandleCertificateResponseV3 {
+                            effects: signed_effects.into_inner(),
+                            events: response.events,
+                            input_objects: response.input_objects,
+                            output_objects: response.output_objects,
+                            auxiliary_data: response.auxiliary_data,
+                        })
+                    })
+                    .collect::<Result<Vec<HandleCertificateResponseV3>, tonic::Status>>()?,
+            )
+        } else {
+            None
+        };
+
+        Ok((responses, weight))
+    }
 
+    async fn handle_submit_to_consensus(
+        &self,
+        consensus_transactions: NonEmpty<ConsensusTransaction>,
+        include_events: bool,
+        include_input_objects: bool,
+        include_output_objects: bool,
+        _include_auxiliary_data: bool,
+        epoch_store: &Arc<AuthorityPerEpochStore>,
+        wait_for_effects: bool,
+    ) -> Result<(Option<Vec<HandleTransactionResponseV2>>, Weight), tonic::Status> {
+        let consensus_transactions: Vec<_> = consensus_transactions.into();
         {
             // code block within reconfiguration lock
             let reconfiguration_lock = epoch_store.get_reconfig_state_read_lock_guard();
@@ -565,29 +762,27 @@ impl ValidatorService {
                 return Err(SuiError::ValidatorHaltedAtEpochEnd.into());
             }
 
-            // 3) All certificates are sent to consensus (at least by some authorities)
-            // For shared objects this will wait until either timeout or we have heard back from consensus.
-            // For owned objects this will return without waiting for certificate to be sequenced
+            // 3) All transactions are sent to consensus (at least by some authorities)
+            // For certs with shared objects this will wait until either timeout or we have heard back from consensus.
+            // For certs with owned objects this will return without waiting for certificate to be sequenced.
+            // For uncertified transactions this will wait for fast path processing.
             // First do quick dirty non-async check.
-            if !epoch_store
-                .is_all_tx_certs_consensus_message_processed(verified_certificates.iter())?
-            {
-                let _metrics_guard = if shared_object_tx {
+            if !epoch_store.all_external_consensus_messages_processed(
+                consensus_transactions.iter().map(|tx| tx.key()),
+            )? {
+                let _metrics_guard = if consensus_transactions.iter().any(|tx| match &tx.kind {
+                    ConsensusTransactionKind::CertifiedTransaction(tx) => {
+                        tx.contains_shared_object()
+                    }
+                    ConsensusTransactionKind::UserTransaction(_) => true,
+                    _ => false,
+                }) {
                     Some(self.metrics.consensus_latency.start_timer())
                 } else {
                     None
                 };
-                let transactions = verified_certificates
-                    .iter()
-                    .map(|certificate| {
-                        ConsensusTransaction::new_certificate_message(
-                            &self.state.name,
-                            certificate.clone().into(),
-                        )
-                    })
-                    .collect::<Vec<_>>();
                 self.consensus_adapter.submit_batch(
-                    &transactions,
+                    &consensus_transactions,
                     Some(&reconfiguration_lock),
                     epoch_store,
                 )?;
@@ -599,10 +794,17 @@ impl ValidatorService {
         if !wait_for_effects {
             // It is useful to enqueue owned object transaction for execution locally,
             // even when we are not returning effects to user
-            let certificates_without_shared_objects = verified_certificates
+            let certificates_without_shared_objects = consensus_transactions
                 .iter()
-                .filter(|certificate| !certificate.contains_shared_object())
-                .cloned()
+                .filter_map(|tx| {
+                    if let ConsensusTransactionKind::CertifiedTransaction(certificate) = &tx.kind {
+                        (!certificate.contains_shared_object())
+                            // Certificates already verified by callers of this function.
+                            .then_some(VerifiedCertificate::new_unchecked(*(certificate.clone())))
+                    } else {
+                        None
+                    }
+                })
                 .collect::<Vec<_>>();
             if !certificates_without_shared_objects.is_empty() {
                 self.state.enqueue_certificates_for_execution(
@@ -615,12 +817,21 @@ impl ValidatorService {
 
         // 4) Execute the certificates immediately if they contain only owned object transactions,
         // or wait for the execution results if it contains shared objects.
-        let responses = futures::future::try_join_all(verified_certificates.into_iter().map(
-            |certificate| async move {
-                let effects = self
-                    .state
-                    .execute_certificate(&certificate, epoch_store)
-                    .await?;
+        let responses = futures::future::try_join_all(consensus_transactions.into_iter().map(
+            |tx| async move {
+                let effects = match &tx.kind {
+                    ConsensusTransactionKind::CertifiedTransaction(certificate) => {
+                        // Certificates already verified by callers of this function.
+                        let certificate = VerifiedCertificate::new_unchecked(*(certificate.clone()));
+                         self.state
+                            .execute_certificate(&certificate, epoch_store)
+                            .await?
+                    }
+                    ConsensusTransactionKind::UserTransaction(tx) => {
+                        self.state.await_transaction_effects(*tx.digest()).await?
+                    }
+                    _ => panic!("`handle_submit_to_consensus` received transaction that is not a CertifiedTransaction or UserTransaction"),
+                };
                 let events = if include_events {
                     if let Some(digest) = effects.events_digest() {
                         Some(self.state.get_transaction_events(digest)?)
@@ -639,11 +850,13 @@ impl ValidatorService {
                     .then(|| self.state.get_transaction_output_objects(&effects))
                     .and_then(Result::ok);
 
-                let signed_effects = self.state.sign_effects(effects, epoch_store)?;
-                epoch_store.insert_tx_cert_sig(certificate.digest(), certificate.auth_sig())?;
+                if let ConsensusTransactionKind::CertifiedTransaction(certificate) = &tx.kind {
+                    epoch_store.insert_tx_cert_sig(certificate.digest(), certificate.auth_sig())?;
+                    // TODO(fastpath): Make sure consensus handler does this for a UserTransaction.
+                }
 
-                Ok::<_, SuiError>(HandleCertificateResponseV3 {
-                    effects: signed_effects.into_inner(),
+                Ok::<_, SuiError>(HandleTransactionResponseV2 {
+                    effects,
                     events,
                     input_objects,
                     output_objects,
@@ -667,6 +880,13 @@ impl ValidatorService {
         self.handle_transaction(request).await
     }
 
+    async fn transaction_v2_impl(
+        &self,
+        request: tonic::Request<HandleTransactionRequestV2>,
+    ) -> WrappedServiceResponse<HandleTransactionResponseV2> {
+        self.handle_transaction_v2(request).await
+    }
+
     async fn submit_certificate_impl(
         &self,
         request: tonic::Request<CertifiedTransaction>,
@@ -1144,6 +1364,23 @@ impl Validator for ValidatorService {
         .unwrap()
     }
 
+    async fn transaction_v2(
+        &self,
+        request: tonic::Request<HandleTransactionRequestV2>,
+    ) -> Result<tonic::Response<HandleTransactionResponseV2>, tonic::Status> {
+        let validator_service = self.clone();
+
+        // Spawns a task which handles the transaction. The task will unconditionally continue
+        // processing in the event that the client connection is dropped.
+        spawn_monitored_task!(async move {
+            // NB: traffic tally wrapping handled within the task rather than on task exit
+            // to prevent an attacker from subverting traffic control by severing the connection
+            handle_with_decoration!(validator_service, transaction_v2_impl, request)
+        })
+        .await
+        .unwrap()
+    }
+
     async fn submit_certificate(
         &self,
         request: tonic::Request<CertifiedTransaction>,
diff --git a/crates/sui-core/src/consensus_adapter.rs b/crates/sui-core/src/consensus_adapter.rs
index 20ff98da8180d..acf7f550bc6e6 100644
--- a/crates/sui-core/src/consensus_adapter.rs
+++ b/crates/sui-core/src/consensus_adapter.rs
@@ -361,6 +361,12 @@ impl ConsensusAdapter {
         committee: &Committee,
         transactions: &[ConsensusTransaction],
     ) -> (impl Future<Output = ()>, usize, usize, usize) {
+        if transactions.iter().any(|tx| tx.is_user_transaction()) {
+            // UserTransactions are generally sent to just one validator and should
+            // be submitted to consensus without delay.
+            return (tokio::time::sleep(Duration::ZERO), 0, 0, 0);
+        }
+
         // Use the minimum digest to compute submit delay.
         let min_digest = transactions
             .iter()
@@ -580,8 +586,8 @@ impl ConsensusAdapter {
         epoch_store: &Arc<AuthorityPerEpochStore>,
     ) -> SuiResult<JoinHandle<()>> {
         if transactions.len() > 1 {
-            // In soft bundle, we need to check if all transactions are of UserTransaction kind.
-            // The check is required because we assume this in submit_and_wait_inner.
+            // In soft bundle, we need to check if all transactions are of CertifiedTransaction
+            // kind. The check is required because we assume this in submit_and_wait_inner.
             for transaction in transactions {
                 fp_ensure!(
                     matches!(
diff --git a/crates/sui-core/src/consensus_handler.rs b/crates/sui-core/src/consensus_handler.rs
index d2fc91649fbbc..d3634d53711e9 100644
--- a/crates/sui-core/src/consensus_handler.rs
+++ b/crates/sui-core/src/consensus_handler.rs
@@ -668,7 +668,7 @@ impl SequencedConsensusTransactionKind {
 
     pub fn is_executable_transaction(&self) -> bool {
         match self {
-            SequencedConsensusTransactionKind::External(ext) => ext.is_user_certificate(),
+            SequencedConsensusTransactionKind::External(ext) => ext.is_certified_transaction(),
             SequencedConsensusTransactionKind::System(_) => true,
         }
     }
diff --git a/crates/sui-core/src/test_authority_clients.rs b/crates/sui-core/src/test_authority_clients.rs
index a1480050e1c1f..f337f564673bc 100644
--- a/crates/sui-core/src/test_authority_clients.rs
+++ b/crates/sui-core/src/test_authority_clients.rs
@@ -219,7 +219,7 @@ impl LocalAuthorityClient {
                     .await?;
                 //let certificate = certificate.verify(epoch_store.committee())?;
                 state.enqueue_certificates_for_execution(vec![certificate.clone()], &epoch_store);
-                let effects = state.notify_read_effects(&certificate).await?;
+                let effects = state.notify_read_effects(*certificate.digest()).await?;
                 state.sign_effects(effects, &epoch_store)?
             }
         }
diff --git a/crates/sui-core/src/unit_tests/authority_tests.rs b/crates/sui-core/src/unit_tests/authority_tests.rs
index 3ea94cb10bfe7..cda49efeab040 100644
--- a/crates/sui-core/src/unit_tests/authority_tests.rs
+++ b/crates/sui-core/src/unit_tests/authority_tests.rs
@@ -4693,7 +4693,10 @@ async fn test_shared_object_transaction_ok() {
     authority.try_execute_for_test(&certificate).await.unwrap();
 
     // Ensure transaction effects are available.
-    authority.notify_read_effects(&certificate).await.unwrap();
+    authority
+        .notify_read_effects(*certificate.digest())
+        .await
+        .unwrap();
 
     // Ensure shared object sequence number increased.
     let shared_object_version = authority
diff --git a/crates/sui-network/build.rs b/crates/sui-network/build.rs
index 846fb75130f22..f078af2bb2a2b 100644
--- a/crates/sui-network/build.rs
+++ b/crates/sui-network/build.rs
@@ -31,6 +31,15 @@ fn main() -> Result<()> {
                 .codec_path(codec_path)
                 .build(),
         )
+        .method(
+            Method::builder()
+                .name("transaction_v2")
+                .route_name("TransactionV2")
+                .input_type("sui_types::messages_grpc::HandleTransactionRequestV2")
+                .output_type("sui_types::messages_grpc::HandleTransactionResponseV2")
+                .codec_path(codec_path)
+                .build(),
+        )
         .method(
             Method::builder()
                 .name("handle_certificate_v2")
diff --git a/crates/sui-open-rpc/spec/openrpc.json b/crates/sui-open-rpc/spec/openrpc.json
index f6c6f1b73546d..dec4af58bade0 100644
--- a/crates/sui-open-rpc/spec/openrpc.json
+++ b/crates/sui-open-rpc/spec/openrpc.json
@@ -1326,6 +1326,7 @@
                 "loaded_child_object_format_type": false,
                 "loaded_child_objects_fixed": true,
                 "missing_type_is_compatibility_error": true,
+                "mysticeti_fastpath": false,
                 "mysticeti_leader_scoring_and_schedule": false,
                 "mysticeti_use_committed_subdag_digest": false,
                 "narwhal_certificate_v2": false,
diff --git a/crates/sui-protocol-config/src/lib.rs b/crates/sui-protocol-config/src/lib.rs
index faa0fcfbc0592..6160be8d541df 100644
--- a/crates/sui-protocol-config/src/lib.rs
+++ b/crates/sui-protocol-config/src/lib.rs
@@ -182,6 +182,7 @@ const MAX_PROTOCOL_VERSION: u64 = 61;
 //             Enable configuration of maximum number of type nodes in a type layout.
 // Version 61: Switch to distributed vote scoring in consensus in testnet
 //             Further reduce minimum number of random beacon shares.
+//             Add feature flag for Mysticeti fastpath.
 
 #[derive(Copy, Clone, Debug, Hash, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ProtocolVersion(u64);
@@ -536,6 +537,10 @@ struct FeatureFlags {
     // Validate identifier inputs separately
     #[serde(skip_serializing_if = "is_false")]
     validate_identifier_inputs: bool,
+
+    // Enables Mysticeti fastpath.
+    #[serde(skip_serializing_if = "is_false")]
+    mysticeti_fastpath: bool,
 }
 
 fn is_false(b: &bool) -> bool {
@@ -1602,9 +1607,14 @@ impl ProtocolConfig {
     pub fn validate_identifier_inputs(&self) -> bool {
         self.feature_flags.validate_identifier_inputs
     }
+
     pub fn gc_depth(&self) -> u32 {
         self.consensus_gc_depth.unwrap_or(0)
     }
+
+    pub fn mysticeti_fastpath(&self) -> bool {
+        self.feature_flags.mysticeti_fastpath
+    }
 }
 
 #[cfg(not(msim))]
@@ -2792,6 +2802,11 @@ impl ProtocolConfig {
                     }
                     // Further reduce minimum number of random beacon shares.
                     cfg.random_beacon_reduction_lower_bound = Some(700);
+
+                    if chain != Chain::Mainnet && chain != Chain::Testnet {
+                        // Enable Mysticeti fastpath for devnet
+                        cfg.feature_flags.mysticeti_fastpath = true;
+                    }
                 }
                 // Use this template when making changes:
                 //
diff --git a/crates/sui-protocol-config/src/snapshots/sui_protocol_config__test__version_61.snap b/crates/sui-protocol-config/src/snapshots/sui_protocol_config__test__version_61.snap
index b20d3b7505301..a386a9bcc1ab5 100644
--- a/crates/sui-protocol-config/src/snapshots/sui_protocol_config__test__version_61.snap
+++ b/crates/sui-protocol-config/src/snapshots/sui_protocol_config__test__version_61.snap
@@ -68,6 +68,7 @@ feature_flags:
   consensus_distributed_vote_scoring_strategy: true
   consensus_round_prober: true
   validate_identifier_inputs: true
+  mysticeti_fastpath: true
 max_tx_size_bytes: 131072
 max_input_objects: 2048
 max_size_written_objects: 5000000
diff --git a/crates/sui-types/src/messages_consensus.rs b/crates/sui-types/src/messages_consensus.rs
index 7b82c4a0345e9..4a192ab534595 100644
--- a/crates/sui-types/src/messages_consensus.rs
+++ b/crates/sui-types/src/messages_consensus.rs
@@ -393,6 +393,18 @@ impl ConsensusTransaction {
         }
     }
 
+    pub fn new_user_transaction_message(authority: &AuthorityName, tx: Transaction) -> Self {
+        let mut hasher = DefaultHasher::new();
+        let tx_digest = tx.digest();
+        tx_digest.hash(&mut hasher);
+        authority.hash(&mut hasher);
+        let tracking_id = hasher.finish().to_le_bytes();
+        Self {
+            tracking_id,
+            kind: ConsensusTransactionKind::UserTransaction(Box::new(tx)),
+        }
+    }
+
     pub fn new_checkpoint_signature_message(data: CheckpointSignatureMessage) -> Self {
         let mut hasher = DefaultHasher::new();
         data.summary.auth_sig().signature.hash(&mut hasher);
@@ -537,10 +549,14 @@ impl ConsensusTransaction {
         }
     }
 
-    pub fn is_user_certificate(&self) -> bool {
+    pub fn is_certified_transaction(&self) -> bool {
         matches!(self.kind, ConsensusTransactionKind::CertifiedTransaction(_))
     }
 
+    pub fn is_user_transaction(&self) -> bool {
+        matches!(self.kind, ConsensusTransactionKind::UserTransaction(_))
+    }
+
     pub fn is_end_of_publish(&self) -> bool {
         matches!(self.kind, ConsensusTransactionKind::EndOfPublish(_))
     }
diff --git a/crates/sui-types/src/messages_grpc.rs b/crates/sui-types/src/messages_grpc.rs
index ffe8001980e68..1d2ce63d32e51 100644
--- a/crates/sui-types/src/messages_grpc.rs
+++ b/crates/sui-types/src/messages_grpc.rs
@@ -4,10 +4,11 @@
 use crate::base_types::{ObjectID, SequenceNumber, TransactionDigest};
 use crate::crypto::{AuthoritySignInfo, AuthorityStrongQuorumSignInfo};
 use crate::effects::{
-    SignedTransactionEffects, TransactionEvents, VerifiedSignedTransactionEffects,
+    SignedTransactionEffects, TransactionEffects, TransactionEvents,
+    VerifiedSignedTransactionEffects,
 };
 use crate::object::Object;
-use crate::transaction::{CertifiedTransaction, SenderSignedData, SignedTransaction};
+use crate::transaction::{CertifiedTransaction, SenderSignedData, SignedTransaction, Transaction};
 use move_core_types::annotated_value::MoveStructLayout;
 use serde::{Deserialize, Serialize};
 
@@ -184,7 +185,7 @@ pub struct SystemStateRequest {
 
 /// Response type for version 3 of the handle certifacte validator API.
 ///
-/// The coorisponding version 3 request type allows for a client to request events as well as
+/// The corresponding version 3 request type allows for a client to request events as well as
 /// input/output objects from a transaction's execution. Given Validators operate with very
 /// aggressive object pruning, the return of input/output objects is only done immediately after
 /// the transaction has been executed locally on the validator and will not be returned for
@@ -219,6 +220,43 @@ pub struct HandleCertificateRequestV3 {
     pub include_auxiliary_data: bool,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct HandleTransactionRequestV2 {
+    pub transaction: Transaction,
+
+    pub include_events: bool,
+    pub include_input_objects: bool,
+    pub include_output_objects: bool,
+    pub include_auxiliary_data: bool,
+}
+
+/// Response type for version 2 of the handle transaction validator API.
+///
+/// The corresponding version 2 request type allows for a client to request events as well as
+/// input/output objects from a transaction's execution. Given Validators operate with very
+/// aggressive object pruning, the return of input/output objects is only done immediately after
+/// the transaction has been executed locally on the validator and will not be returned for
+/// requests to previously executed transactions.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct HandleTransactionResponseV2 {
+    pub effects: TransactionEffects,
+    pub events: Option<TransactionEvents>,
+
+    /// If requested, will included all initial versions of objects modified in this transaction.
+    /// This includes owned objects included as input into the transaction as well as the assigned
+    /// versions of shared objects.
+    //
+    // TODO: In the future we may want to include shared objects or child objects which were read
+    // but not modified during exectuion.
+    pub input_objects: Option<Vec<Object>>,
+
+    /// If requested, will included all changed objects, including mutated, created and unwrapped
+    /// objects. In other words, all objects that still exist in the object state after this
+    /// transaction.
+    pub output_objects: Option<Vec<Object>>,
+    pub auxiliary_data: Option<Vec<u8>>,
+}
+
 impl From<HandleCertificateResponseV3> for HandleCertificateResponseV2 {
     fn from(value: HandleCertificateResponseV3) -> Self {
         Self {