Update ConsensusTransactionKind and ConsensusTransactionKey usages

crates/sui-benchmark/src/lib.rs

@@ -45,7 +45,7 @@ use sui_types::{
     sui_system_state::SuiSystemStateTrait,
 };
 use tokio::time::sleep;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 pub mod bank;
 pub mod benchmark_setup;
@@ -375,28 +375,17 @@ impl ValidatorProxy for LocalValidatorAggregatorProxy {
                         events,
                         ..
                     } = resp;
-                    error!(
-                        ?tx_digest,
-                        retry_cnt,
-                        "Transaction succeeded",
-                    );
                     return Ok(ExecutionEffects::FinalizedTransactionEffects(
                         FinalizedEffects::new_from_effects_cert(effects_cert.into()),
                         events.unwrap_or_default(),
                     ));
                 }
                 Err(QuorumDriverError::NonRecoverableTransactionError { errors }) => {
-                    error!(
-                        ?tx_digest,
-                        retry_cnt,
-                        "Transaction failed with non-recoverable err: {:?}",
-                        errors
-                    );
                     bail!(QuorumDriverError::NonRecoverableTransactionError { errors });
                 }
                 Err(err) => {
                     let delay = Duration::from_millis(rand::thread_rng().gen_range(100..1000));
-                    error!(
+                    warn!(
                         ?tx_digest,
                         retry_cnt,
                         "Transaction failed with err: {:?}. Sleeping for {:?} ...",

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

@@ -796,16 +796,24 @@ impl AuthorityEpochTables {
 
         let shared_input_object_ids: BTreeSet<_> = transactions
             .iter()
-            .filter_map(|tx| {
-                if let SequencedConsensusTransactionKind::External(ConsensusTransaction {
+            .filter_map(|tx| match &tx.0.transaction {
+                SequencedConsensusTransactionKind::External(ConsensusTransaction {
                     kind: ConsensusTransactionKind::CertifiedTransaction(tx),
                     ..
-                }) = &tx.0.transaction
-                {
-                    Some(tx.shared_input_objects().map(|obj| obj.id))
-                } else {
-                    None
-                }
+                }) => Some(
+                    tx.shared_input_objects()
+                        .map(|obj| obj.id)
+                        .collect::<Vec<_>>(),
+                ),
+                SequencedConsensusTransactionKind::External(ConsensusTransaction {
+                    kind: ConsensusTransactionKind::UserTransaction(tx),
+                    ..
+                }) => Some(
+                    tx.shared_input_objects()
+                        .map(|obj| obj.id)
+                        .collect::<Vec<_>>(),
+                ),
+                _ => None,
             })
             .flatten()
             .collect();
@@ -859,14 +867,12 @@ impl AuthorityPerEpochStore {
         let pending_consensus_transactions = tables.get_all_pending_consensus_transactions();
         let pending_consensus_certificates: HashSet<_> = pending_consensus_transactions
             .iter()
-            .filter_map(|transaction| {
-                if let ConsensusTransactionKind::CertifiedTransaction(certificate) =
-                    &transaction.kind
-                {
+            .filter_map(|transaction| match &transaction.kind {
+                ConsensusTransactionKind::CertifiedTransaction(certificate) => {
                     Some(*certificate.digest())
-                } else {
-                    None
                 }
+                ConsensusTransactionKind::UserTransaction(txn) => Some(*txn.digest()),
+                _ => None,
             })
             .collect();
         assert_eq!(
@@ -1792,9 +1798,6 @@ impl AuthorityPerEpochStore {
                 txs.iter().map(|tx| match tx.0.transaction.key() {
                     SequencedConsensusTransactionKey::External(
                         ConsensusTransactionKey::Certificate(digest),
-                    )
-                    | SequencedConsensusTransactionKey::External(
-                        ConsensusTransactionKey::UserTransaction(digest),
                     ) => (digest, *deferral_key),
                     _ => {
                         panic!("deferred randomness transaction was not a user certificate: {tx:?}")
@@ -1979,13 +1982,10 @@ impl AuthorityPerEpochStore {
         self.tables()?
             .pending_consensus_transactions
             .multi_remove(keys)?;
-        // TODO: lock once for all remove() calls.
+        let mut pending_consensus_certificates = self.pending_consensus_certificates.write();
         for key in keys {
             if let ConsensusTransactionKey::Certificate(digest) = key {
-                self.pending_consensus_certificates.write().remove(digest);
-            }
-            if let ConsensusTransactionKey::UserTransaction(digest) = key {
-                self.pending_consensus_certificates.write().remove(digest);
+                pending_consensus_certificates.remove(digest);
             }
         }
         Ok(())
@@ -2738,9 +2738,6 @@ impl AuthorityPerEpochStore {
                     txs.iter().map(|tx| match tx.0.transaction.key() {
                         SequencedConsensusTransactionKey::External(
                             ConsensusTransactionKey::Certificate(digest),
-                        )
-                        | SequencedConsensusTransactionKey::External(
-                            ConsensusTransactionKey::UserTransaction(digest),
                         ) => (digest, *deferral_key),
                         _ => panic!("deferred transaction was not a user certificate: {tx:?}"),
                     })

crates/sui-core/src/authority_server.rs

@@ -823,7 +823,7 @@ impl ValidatorService {
                     ConsensusTransactionKind::CertifiedTransaction(certificate) => {
                         // Certificates already verified by callers of this function.
                         let certificate = VerifiedCertificate::new_unchecked(*(certificate.clone()));
-                         self.state
+                        self.state
                             .execute_certificate(&certificate, epoch_store)
                             .await?
                     }

crates/sui-core/src/consensus_adapter.rs

@@ -341,6 +341,10 @@ impl ConsensusAdapter {
                 ConsensusTransactionKind::CertifiedTransaction(certificate) => {
                     Some(certificate.digest())
                 }
+                // TODO(fastpath): support batch of UserTransaction.
+                ConsensusTransactionKind::UserTransaction(transaction) => {
+                    Some(transaction.digest())
+                }
                 _ => None,
             })
             .min();
@@ -563,6 +567,7 @@ impl ConsensusAdapter {
                     ),
                     SuiError::InvalidTxKindInSoftBundle
                 );
+                // TODO(fastpath): support batch of UserTransaction.
             }
         }
 
@@ -804,6 +809,10 @@ impl ConsensusAdapter {
             || matches!(
                 transactions[0].kind,
                 ConsensusTransactionKind::CertifiedTransaction(_)
+            )
+            || matches!(
+                transactions[0].kind,
+                ConsensusTransactionKind::UserTransaction(_)
             );
         let send_end_of_publish = if is_user_tx {
             // If we are in RejectUserCerts state and we just drained the list we need to
@@ -855,9 +864,6 @@ impl ConsensusAdapter {
             let transaction_digests = match transaction_key {
                 SequencedConsensusTransactionKey::External(
                     ConsensusTransactionKey::Certificate(digest),
-                )
-                | SequencedConsensusTransactionKey::External(
-                    ConsensusTransactionKey::UserTransaction(digest),
                 ) => vec![digest],
                 _ => vec![],
             };

crates/sui-core/src/consensus_handler.rs

@@ -653,20 +653,18 @@ impl SequencedConsensusTransactionKind {
 
     pub fn is_executable_transaction(&self) -> bool {
         match self {
-            SequencedConsensusTransactionKind::External(ext) => ext.is_certified_transaction(),
+            SequencedConsensusTransactionKind::External(ext) => ext.is_executable_transaction(),
             SequencedConsensusTransactionKind::System(_) => true,
         }
     }
 
     pub fn executable_transaction_digest(&self) -> Option<TransactionDigest> {
         match self {
-            SequencedConsensusTransactionKind::External(ext) => {
-                if let ConsensusTransactionKind::CertifiedTransaction(txn) = &ext.kind {
-                    Some(*txn.digest())
-                } else {
-                    None
-                }
-            }
+            SequencedConsensusTransactionKind::External(ext) => match &ext.kind {
+                ConsensusTransactionKind::CertifiedTransaction(txn) => Some(*txn.digest()),
+                ConsensusTransactionKind::UserTransaction(txn) => Some(*txn.digest()),
+                _ => None,
+            },
             SequencedConsensusTransactionKind::System(txn) => Some(*txn.digest()),
         }
     }
@@ -711,14 +709,17 @@ impl SequencedConsensusTransaction {
             // which will eventually fail when the randomness state object is not found.
             return false;
         }
-        let SequencedConsensusTransactionKind::External(ConsensusTransaction {
-            kind: ConsensusTransactionKind::CertifiedTransaction(certificate),
-            ..
-        }) = &self.transaction
-        else {
-            return false;
-        };
-        certificate.transaction_data().uses_randomness()
+        match &self.transaction {
+            SequencedConsensusTransactionKind::External(ConsensusTransaction {
+                kind: ConsensusTransactionKind::CertifiedTransaction(cert),
+                ..
+            }) => cert.transaction_data().uses_randomness(),
+            SequencedConsensusTransactionKind::External(ConsensusTransaction {
+                kind: ConsensusTransactionKind::UserTransaction(txn),
+                ..
+            }) => txn.transaction_data().uses_randomness(),
+            _ => false,
+        }
     }
 
     pub fn as_shared_object_txn(&self) -> Option<&SenderSignedData> {
@@ -727,6 +728,10 @@ impl SequencedConsensusTransaction {
                 kind: ConsensusTransactionKind::CertifiedTransaction(certificate),
                 ..
             }) if certificate.contains_shared_object() => Some(certificate.data()),
+            SequencedConsensusTransactionKind::External(ConsensusTransaction {
+                kind: ConsensusTransactionKind::UserTransaction(txn),
+                ..
+            }) if txn.contains_shared_object() => Some(txn.data()),
             SequencedConsensusTransactionKind::System(txn) if txn.contains_shared_object() => {
                 Some(txn.data())
             }
@@ -1390,6 +1395,9 @@ mod tests {
                 ConsensusTransactionKind::CertifiedTransaction(txn) => {
                     format!("user({})", txn.transaction_data().gas_price())
                 }
+                ConsensusTransactionKind::UserTransaction(txn) => {
+                    format!("user({})", txn.transaction_data().gas_price())
+                }
                 _ => unreachable!(),
             },
             SequencedConsensusTransactionKind::System(_) => unreachable!(),

crates/sui-core/src/mock_consensus.rs

@@ -9,8 +9,9 @@ use crate::consensus_handler::SequencedConsensusTransaction;
 use prometheus::Registry;
 use std::sync::{Arc, Weak};
 use sui_types::error::{SuiError, SuiResult};
+use sui_types::executable_transaction::VerifiedExecutableTransaction;
 use sui_types::messages_consensus::{ConsensusTransaction, ConsensusTransactionKind};
-use sui_types::transaction::VerifiedCertificate;
+use sui_types::transaction::{VerifiedCertificate, VerifiedTransaction};
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tracing::debug;
@@ -73,10 +74,21 @@ impl MockConsensusClient {
                         .unwrap();
                 }
             }
-            if let ConsensusTransactionKind::CertifiedTransaction(tx) = tx.kind {
+            if let ConsensusTransactionKind::CertifiedTransaction(tx) = &tx.kind {
                 if tx.contains_shared_object() {
                     validator.enqueue_certificates_for_execution(
-                        vec![VerifiedCertificate::new_unchecked(*tx)],
+                        vec![VerifiedCertificate::new_unchecked(*tx.clone())],
+                        &epoch_store,
+                    );
+                }
+            }
+            if let ConsensusTransactionKind::UserTransaction(tx) = &tx.kind {
+                if tx.contains_shared_object() {
+                    validator.enqueue_transactions_for_execution(
+                        vec![VerifiedExecutableTransaction::new_from_consensus(
+                            VerifiedTransaction::new_unchecked(*tx.clone()),
+                            0,
+                        )],
                         &epoch_store,
                     );
                 }

crates/sui-core/src/post_consensus_tx_reorder.rs

@@ -6,7 +6,10 @@ use crate::consensus_handler::{
 };
 use mysten_metrics::monitored_scope;
 use sui_protocol_config::ConsensusTransactionOrdering;
-use sui_types::messages_consensus::{ConsensusTransaction, ConsensusTransactionKind};
+use sui_types::{
+    messages_consensus::{ConsensusTransaction, ConsensusTransactionKind},
+    transaction::TransactionDataAPI as _,
+};
 
 pub struct PostConsensusTxReorder {}
 
@@ -34,6 +37,10 @@ impl PostConsensusTxReorder {
                         tracking_id: _,
                         kind: ConsensusTransactionKind::CertifiedTransaction(cert),
                     }) => cert.gas_price(),
+                    SequencedConsensusTransactionKind::External(ConsensusTransaction {
+                        tracking_id: _,
+                        kind: ConsensusTransactionKind::UserTransaction(txn),
+                    }) => txn.transaction_data().gas_price(),
                     // Non-user transactions are considered to have gas price of MAX u64 and are
                     // put to the beginning.
                     _ => u64::MAX,

crates/sui-types/src/messages_consensus.rs

@@ -111,7 +111,6 @@ pub enum ConsensusTransactionKey {
     NewJWKFetched(Box<(AuthorityName, JwkId, JWK)>),
     RandomnessDkgMessage(AuthorityName),
     RandomnessDkgConfirmation(AuthorityName),
-    UserTransaction(TransactionDigest),
 }
 
 impl Debug for ConsensusTransactionKey {
@@ -144,7 +143,6 @@ impl Debug for ConsensusTransactionKey {
             Self::RandomnessDkgConfirmation(name) => {
                 write!(f, "RandomnessDkgConfirmation({:?})", name.concise())
             }
-            Self::UserTransaction(digest) => write!(f, "UserTransaction({:?})", digest),
         }
     }
 }
@@ -555,13 +553,17 @@ impl ConsensusTransaction {
                 ConsensusTransactionKey::RandomnessDkgConfirmation(*authority)
             }
             ConsensusTransactionKind::UserTransaction(tx) => {
-                ConsensusTransactionKey::UserTransaction(*tx.digest())
+                // Use the same key format as ConsensusTransactionKind::CertifiedTransaction,
+                // because existing usages of ConsensusTransactionKey should not differentiate
+                // between CertifiedTransaction and UserTransaction.
+                ConsensusTransactionKey::Certificate(*tx.digest())
             }
         }
     }
 
-    pub fn is_certified_transaction(&self) -> bool {
+    pub fn is_executable_transaction(&self) -> bool {
         matches!(self.kind, ConsensusTransactionKind::CertifiedTransaction(_))
+            || matches!(self.kind, ConsensusTransactionKind::UserTransaction(_))
     }
 
     pub fn is_user_transaction(&self) -> bool {