[mfp] log actions during transaction submission (#23846)

## Description 

Log transaction submission steps with timestamps, for debugging and
latency investigations.

## Test plan 

CI

Author: mwtian

crates/sui-core/src/authority_server.rs

@@ -6,6 +6,7 @@ use anyhow::Result;
 use async_trait::async_trait;
 use fastcrypto::traits::KeyPair;
 use futures::{future, TryFutureExt};
+use itertools::Itertools as _;
 use mysten_metrics::spawn_monitored_task;
 use prometheus::{
     register_gauge_with_registry, register_histogram_vec_with_registry,
@@ -566,7 +567,12 @@ impl ValidatorService {
         Ok((tonic::Response::new(info), Weight::zero()))
     }
 
-    #[instrument(name= "ValidatorService::handle_submit_transaction", level = "error", skip_all, err(level = "debug"), fields(tx_digest = ?tracing::field::Empty))]
+    #[instrument(
+        name = "ValidatorService::handle_submit_transaction",
+        level = "error",
+        skip_all,
+        err(level = "debug")
+    )]
     async fn handle_submit_transaction(
         &self,
         request: tonic::Request<RawSubmitTxRequest>,
@@ -644,6 +650,8 @@ impl ValidatorService {
             .into()
         );
 
+        // Transaction digests.
+        let mut tx_digests = Vec::with_capacity(request.transactions.len());
         // Transactions to submit to consensus.
         let mut consensus_transactions = Vec::with_capacity(request.transactions.len());
         // Indexes of transactions above in the request transactions.
@@ -714,11 +722,12 @@ impl ValidatorService {
             };
 
             let tx_digest = verified_transaction.digest();
+            tx_digests.push(*tx_digest);
 
-            // Only trace the 1st transaction in the request.
-            if idx == 0 {
-                tracing::Span::current().record("tx_digest", tracing::field::debug(&tx_digest));
-            }
+            debug!(
+                ?tx_digest,
+                "handle_submit_transaction: verified transaction"
+            );
 
             // Check if the transaction has executed, before checking input objects
             // which could have been consumed.
@@ -744,17 +753,22 @@ impl ValidatorService {
                         fast_path: false,
                     };
                     results[idx] = Some(executed_result);
+                    debug!(?tx_digest, "handle_submit_transaction: already executed");
                     continue;
                 }
             }
 
+            debug!(
+                ?tx_digest,
+                "handle_submit_transaction: waiting for fastpath dependency objects"
+            );
             if !state
                 .wait_for_fastpath_dependency_objects(&verified_transaction, epoch_store.epoch())
                 .await?
             {
                 debug!(
                     ?tx_digest,
-                    "Fastpath input objects are still unavailable after waiting"
+                    "fastpath input objects are still unavailable after waiting"
                 );
             }
 
@@ -859,6 +873,14 @@ impl ValidatorService {
                 is_ping_request || !consensus_transactions.is_empty(),
                 "A valid soft bundle must have at least one transaction"
             );
+            debug!(
+                "handle_submit_transaction: submitting consensus transactions ({}): {}",
+                req_type,
+                consensus_transactions
+                    .iter()
+                    .map(|t| t.local_display())
+                    .join(", ")
+            );
             self.handle_submit_to_consensus_for_position(
                 consensus_transactions,
                 &epoch_store,
@@ -867,6 +889,11 @@ impl ValidatorService {
             .await?
         } else {
             let futures = consensus_transactions.into_iter().map(|t| {
+                debug!(
+                    "handle_submit_transaction: submitting consensus transaction ({}): {}",
+                    req_type,
+                    t.local_display(),
+                );
                 self.handle_submit_to_consensus_for_position(
                     vec![t],
                     &epoch_store,
@@ -888,9 +915,16 @@ impl ValidatorService {
             }));
         } else {
             // Otherwise, return the consensus position for each transaction.
-            for (idx, consensus_position) in
-                transaction_indexes.into_iter().zip(consensus_positions)
+            for ((idx, tx_digest), consensus_position) in transaction_indexes
+                .into_iter()
+                .zip(tx_digests)
+                .zip(consensus_positions)
             {
+                debug!(
+                    ?tx_digest,
+                    "handle_submit_transaction: submitted consensus transaction at {}",
+                    consensus_position,
+                );
                 results[idx] = Some(SubmitTxResult::Submitted { consensus_position });
             }
         }
@@ -965,6 +999,7 @@ impl ValidatorService {
         //    When multiple certificates are provided, we will either submit all of them or none of them to consensus.
         if certificates.len() == 1 {
             let tx_digest = *certificates[0].digest();
+            debug!(tx_digest=?tx_digest, "Checking if certificate is already executed");
 
             if let Some(signed_effects) = self
                 .state
@@ -1075,7 +1110,7 @@ impl ValidatorService {
         name = "ValidatorService::handle_submit_to_consensus_for_position",
         level = "debug",
         skip_all,
-        err
+        err(level = "debug")
     )]
     async fn handle_submit_to_consensus_for_position(
         &self,
@@ -1418,7 +1453,7 @@ impl ValidatorService {
         request: WaitForEffectsRequest,
         epoch_store: &Arc<AuthorityPerEpochStore>,
     ) -> SuiResult<WaitForEffectsResponse> {
-        if request.ping.is_some() {
+        if request.ping_type.is_some() {
             return timeout(
                 Duration::from_secs(10),
                 self.ping_response(request, epoch_store),
@@ -1502,7 +1537,7 @@ impl ValidatorService {
         };
 
         // We assume that the caller has already checked for the existence of the `ping` field, but handling it gracefully here.
-        let Some(ping) = request.ping else {
+        let Some(ping) = request.ping_type else {
             return Err(SuiError::InvalidRequest(
                 "Ping type is required for ping requests".to_string(),
             ));

crates/sui-core/src/transaction_driver/effects_certifier.rs

@@ -70,7 +70,6 @@ impl EffectsCertifier {
         // Guaranteed to be not the Rejected variant.
         submit_txn_result: SubmitTxResult,
         options: &SubmitTransactionOptions,
-        ping: Option<PingType>,
     ) -> Result<QuorumTransactionResponse, TransactionDriverError>
     where
         A: AuthorityAPI + Send + Sync + 'static + Clone,
@@ -117,7 +116,6 @@ impl EffectsCertifier {
                 consensus_position,
                 options,
                 current_target,
-                ping
             ),
             async {
                 // No need to send a full effects request if it is already provided.
@@ -132,7 +130,7 @@ impl EffectsCertifier {
                     .expect("there should be at least 1 target");
                 current_target = name;
                 full_effects_start_time = Some(Instant::now());
-                self.get_full_effects(client, tx_digest, consensus_position, options, ping)
+                self.get_full_effects(client, tx_digest, tx_type, consensus_position, options)
                     .await
             },
         );
@@ -200,7 +198,7 @@ impl EffectsCertifier {
             current_target = name;
             full_effects_start_time = Some(Instant::now());
             full_effects_result = self
-                .get_full_effects(client, tx_digest, consensus_position, options, ping)
+                .get_full_effects(client, tx_digest, tx_type, consensus_position, options)
                 .await;
         }
     }
@@ -210,18 +208,19 @@ impl EffectsCertifier {
         &self,
         client: Arc<SafeClient<A>>,
         tx_digest: Option<TransactionDigest>,
+        tx_type: TxType,
         consensus_position: Option<ConsensusPosition>,
         options: &SubmitTransactionOptions,
-        ping: Option<PingType>,
     ) -> Result<(TransactionEffectsDigest, Box<ExecutedData>, bool), TransactionRequestError>
     where
         A: AuthorityAPI + Send + Sync + 'static + Clone,
     {
+        let ping_type = get_ping_type(&tx_digest, tx_type);
         let request = WaitForEffectsRequest {
             transaction_digest: tx_digest,
             consensus_position,
             include_details: true,
-            ping,
+            ping_type,
         };
 
         match timeout(
@@ -272,12 +271,12 @@ impl EffectsCertifier {
         consensus_position: Option<ConsensusPosition>,
         options: &SubmitTransactionOptions,
         submitted_tx_to_validator: AuthorityName,
-        ping: Option<PingType>,
     ) -> Result<TransactionEffectsDigest, TransactionDriverError>
     where
         A: AuthorityAPI + Send + Sync + 'static + Clone,
     {
-        let ping_label = if ping.is_some() { "true" } else { "false" };
+        let ping_type = get_ping_type(&tx_digest, tx_type);
+        let ping_label = if tx_digest.is_none() { "true" } else { "false" };
         self.metrics
             .certified_effects_ack_attempts
             .with_label_values(&[tx_type.as_str(), ping_label])
@@ -292,7 +291,7 @@ impl EffectsCertifier {
             transaction_digest: tx_digest,
             consensus_position,
             include_details: false,
-            ping,
+            ping_type,
         })
         .unwrap();
 
@@ -650,3 +649,14 @@ impl EffectsCertifier {
         }
     }
 }
+
+fn get_ping_type(tx_digest: &Option<TransactionDigest>, tx_type: TxType) -> Option<PingType> {
+    if tx_digest.is_none() {
+        Some(match tx_type {
+            TxType::SingleWriter => PingType::FastPath,
+            TxType::SharedObject => PingType::Consensus,
+        })
+    } else {
+        None
+    }
+}

crates/sui-core/src/transaction_driver/mod.rs

@@ -178,7 +178,7 @@ where
                 }
                 let start_time = Instant::now();
 
-                let ping = if tx_type == TxType::SingleWriter {
+                let ping_type = if tx_type == TxType::SingleWriter {
                     PingType::FastPath
                 } else {
                     PingType::Consensus
@@ -187,7 +187,7 @@ where
                 // Now send a ping transaction to the chosen validator for the provided tx type
                 match self_clone
                     .drive_transaction(
-                        SubmitTxRequest::new_ping(ping),
+                        SubmitTxRequest::new_ping(ping_type),
                         SubmitTransactionOptions {
                             allowed_validators: vec![name],
                             ..Default::default()
@@ -214,16 +214,16 @@ where
         }
     }
 
-    /// Drives transaction to submission and effects certification. If ping is provided, then the requested will be treated as a ping transaction.
-    #[instrument(level = "error", skip_all, fields(tx_digest = ?request.transaction.as_ref().map(|t| t.digest()), ping = ?request.ping))]
+    /// Drives transaction to submission and effects certification. Ping requests are derived from the submitted payload.
+    #[instrument(level = "error", skip_all, fields(tx_digest = ?request.transaction.as_ref().map(|t| t.digest()), ping = %request.ping_type.is_some()))]
     pub async fn drive_transaction(
         &self,
         request: SubmitTxRequest,
         options: SubmitTransactionOptions,
         timeout_duration: Option<Duration>,
     ) -> Result<QuorumTransactionResponse, TransactionDriverError> {
         // For ping requests, the amplification factor is always 1.
-        let amplification_factor = if request.ping.is_some() {
+        let amplification_factor = if request.ping_type.is_some() {
             1
         } else {
             let gas_price = request
@@ -247,7 +247,7 @@ where
         };
 
         let tx_type = request.tx_type();
-        let ping_label = if request.ping.is_some() {
+        let ping_label = if request.ping_type.is_some() {
             "true"
         } else {
             "false"
@@ -358,7 +358,6 @@ where
         let amplification_factor =
             amplification_factor.min(auth_agg.committee.num_members() as u64);
         let start_time = Instant::now();
-        let ping = request.ping;
         let tx_type = request.tx_type();
         let tx_digest = request.tx_digest();
 
@@ -390,7 +389,6 @@ where
                 name,
                 submit_txn_result,
                 options,
-                ping,
             )
             .await;
 

crates/sui-core/src/transaction_driver/transaction_submitter.rs

@@ -73,7 +73,7 @@ impl TransactionSubmitter {
             options.allowed_validators.clone(),
         );
 
-        let ping_label = if request.ping.is_some() {
+        let ping_label = if request.ping_type.is_some() {
             "true"
         } else {
             "false"