apollo_transaction_converter: return a verification handle when (#12239)

converting transactions

Author: einat-starkware

crates/apollo_consensus_orchestrator/src/test_utils.rs

@@ -102,8 +102,11 @@ pub(crate) static INTERNAL_TX_BATCH: LazyLock<Vec<InternalConsensusTransaction>>
             .iter()
             .cloned()
             .map(|tx| {
-                block_on(TRANSACTION_CONVERTER.convert_consensus_tx_to_internal_consensus_tx(tx))
-                    .unwrap()
+                let (internal_tx, _) = block_on(
+                    TRANSACTION_CONVERTER.convert_consensus_tx_to_internal_consensus_tx(tx),
+                )
+                .unwrap();
+                internal_tx
             })
             .collect()
     });
@@ -281,7 +284,8 @@ impl TestDeps {
             self.transaction_converter
                 .expect_convert_consensus_tx_to_internal_consensus_tx()
                 .withf(move |internal_tx| internal_tx == tx)
-                .returning(|_| Ok(internal_tx.clone()));
+                // TODO(Einat): Add verification handle when client-side proving is tested in the validate proposal test.
+                .returning(|_| Ok((internal_tx.clone(), None)));
         }
     }
 

crates/apollo_consensus_orchestrator/src/validate_proposal.rs

@@ -20,7 +20,7 @@ use apollo_l1_gas_price_types::L1GasPriceProviderClient;
 use apollo_protobuf::consensus::{ConsensusBlockInfo, ProposalFin, ProposalPart, TransactionBatch};
 use apollo_state_sync_types::communication::StateSyncClient;
 use apollo_time::time::{Clock, ClockExt, DateTime};
-use apollo_transaction_converter::TransactionConverterTrait;
+use apollo_transaction_converter::{TransactionConverterTrait, VerificationHandle};
 use futures::channel::mpsc;
 use futures::StreamExt;
 use starknet_api::block::{BlockNumber, GasPrice};
@@ -414,22 +414,47 @@ async fn handle_proposal_part(
             // TODO(guyn): check that the length of txs and the number of batches we receive is not
             // so big it would fill up the memory (in case of a malicious proposal)
             debug!("Received transaction batch with {} txs", txs.len());
-            let txs =
+            let conversion_results =
                 futures::future::join_all(txs.into_iter().map(|tx| {
                     transaction_converter.convert_consensus_tx_to_internal_consensus_tx(tx)
                 }))
                 .await
                 .into_iter()
                 .collect::<Result<Vec<_>, _>>();
-            let txs = match txs {
-                Ok(txs) => txs,
+            let conversion_results = match conversion_results {
+                Ok(results) => results,
                 Err(e) => {
                     return HandledProposalPart::Failed(format!(
                         "Failed to convert transactions. Stopping the build of the current \
                          proposal. {e:?}"
                     ));
                 }
             };
+
+            // Separate internal transactions from verification handles and collect verification
+            // handles that are not None.
+            let (txs, handles): (
+                Vec<InternalConsensusTransaction>,
+                Vec<Option<VerificationHandle>>,
+            ) = conversion_results.into_iter().unzip();
+            for handle in handles.into_iter().flatten() {
+                let proof_facts_hash = handle.proof_facts.hash();
+                match handle.verification_task.await {
+                    Err(e) => {
+                        return HandledProposalPart::Failed(format!(
+                            "Proof verification task panicked for proof facts hash: \
+                             {proof_facts_hash:?}: {e}"
+                        ));
+                    }
+                    Ok(Err(e)) => {
+                        return HandledProposalPart::Failed(format!(
+                            "Proof verification failed for proof facts hash: \
+                             {proof_facts_hash:?}: {e}"
+                        ));
+                    }
+                    Ok(Ok(())) => {}
+                }
+            }
             debug!(
                 "Converted transactions to internal representation. hashes={:?}",
                 txs.iter().map(|tx| tx.tx_hash()).collect::<Vec<TransactionHash>>()

crates/apollo_gateway/src/gateway.rs

@@ -22,18 +22,21 @@ use apollo_network_types::network_types::BroadcastedMessageMetadata;
 use apollo_proc_macros::sequencer_latency_histogram;
 use apollo_proof_manager_types::SharedProofManagerClient;
 use apollo_state_sync_types::communication::SharedStateSyncClient;
-use apollo_transaction_converter::{TransactionConverter, TransactionConverterTrait};
+use apollo_transaction_converter::{
+    TransactionConverter,
+    TransactionConverterTrait,
+    VerificationHandle,
+};
 use async_trait::async_trait;
 use blockifier::state::contract_class_manager::ContractClassManager;
 use starknet_api::executable_transaction::AccountTransaction;
 use starknet_api::rpc_transaction::{
     InternalRpcTransaction,
     InternalRpcTransactionWithoutTxHash,
     RpcDeclareTransaction,
-    RpcInvokeTransaction,
     RpcTransaction,
 };
-use starknet_api::transaction::fields::TransactionSignature;
+use starknet_api::transaction::fields::{Proof, ProofFacts, TransactionSignature};
 use tracing::{debug, error, info, warn};
 
 use crate::errors::{
@@ -167,18 +170,6 @@ impl<
         let mut metric_counters = GatewayMetricHandle::new(&tx, &p2p_message_metadata);
         metric_counters.count_transaction_received();
 
-        // Extract proof data early for potential archiving later.
-        let proof_data = match tx {
-            RpcTransaction::Invoke(RpcInvokeTransaction::V3(ref tx)) => {
-                if !tx.proof_facts.is_empty() {
-                    Some((tx.proof_facts.clone(), tx.proof.clone()))
-                } else {
-                    None
-                }
-            }
-            _ => None,
-        };
-
         if let RpcTransaction::Declare(ref declare_tx) = tx {
             if let Err(e) = self.check_declare_permissions(declare_tx) {
                 metric_counters.record_add_tx_failure(&e);
@@ -190,7 +181,7 @@ impl<
         self.stateless_tx_validator.validate(&tx)?;
 
         let tx_signature = tx.signature().clone();
-        let (internal_tx, executable_tx) =
+        let (internal_tx, executable_tx, proof_data) =
             self.convert_rpc_tx_to_internal_and_executable_txs(tx, &tx_signature).await?;
 
         let mut stateful_transaction_validator = self
@@ -284,15 +275,24 @@ impl<
         &self,
         tx: RpcTransaction,
         tx_signature: &TransactionSignature,
-    ) -> Result<(InternalRpcTransaction, AccountTransaction), StarknetError> {
-        let internal_tx =
+    ) -> Result<
+        (InternalRpcTransaction, AccountTransaction, Option<(ProofFacts, Proof)>),
+        StarknetError,
+    > {
+        let (internal_tx, verification_handle) =
             self.transaction_converter.convert_rpc_tx_to_internal_rpc_tx(tx).await.map_err(
                 |e| {
                     warn!("Failed to convert RPC transaction to internal RPC transaction: {}", e);
                     transaction_converter_err_to_deprecated_gw_err(tx_signature, e)
                 },
             )?;
 
+        // Extract proof data from verification handle before awaiting to be used later to store the
+        // data.
+        let proof_data = self
+            .await_verification_task_and_extract_proof_data(verification_handle, tx_signature)
+            .await?;
+
         let executable_tx = self
             .transaction_converter
             .convert_internal_rpc_tx_to_executable_tx(internal_tx.clone())
@@ -302,7 +302,29 @@ impl<
                 transaction_converter_err_to_deprecated_gw_err(tx_signature, e)
             })?;
 
-        Ok((internal_tx, executable_tx))
+        Ok((internal_tx, executable_tx, proof_data))
+    }
+    async fn await_verification_task_and_extract_proof_data(
+        &self,
+        verification_handle: Option<VerificationHandle>,
+        tx_signature: &TransactionSignature,
+    ) -> Result<Option<(ProofFacts, Proof)>, StarknetError> {
+        let proof_data = verification_handle
+            .as_ref()
+            .map(|handle| (handle.proof_facts.clone(), handle.proof.clone()));
+
+        // Await the verification task immediately.
+        if let Some(handle) = verification_handle {
+            let verification_result = handle.verification_task.await.map_err(|e| {
+                warn!("Proof verification task join error: {}", e);
+                StarknetError::internal_with_logging("Proof verification task join error:", &e)
+            })?;
+            verification_result.map_err(|e| {
+                warn!("Proof verification failed: {}", e);
+                transaction_converter_err_to_deprecated_gw_err(tx_signature, e)
+            })?;
+        }
+        Ok(proof_data)
     }
 }
 

crates/apollo_gateway/src/gateway_test.rs

@@ -37,6 +37,7 @@ use apollo_transaction_converter::{
     MockTransactionConverterTrait,
     TransactionConverterError,
     TransactionConverterResult,
+    VerificationHandle,
 };
 use blockifier::blockifier::config::ContractClassManagerConfig;
 use blockifier::context::ChainInfo;
@@ -266,11 +267,30 @@ fn setup_transaction_converter_mock(
 ) {
     let rpc_tx = tx_args.get_rpc_tx();
     let internal_tx = tx_args.get_internal_tx();
+
+    // Create verification handle if the transaction has proof facts.
+    let verification_handle =
+        if let RpcTransaction::Invoke(RpcInvokeTransaction::V3(ref invoke_tx)) = rpc_tx {
+            if !invoke_tx.proof_facts.is_empty() {
+                // Create a simple task that just returns Ok (verification is mocked).
+                let verification_task = tokio::spawn(async { Ok(()) });
+                Some(VerificationHandle {
+                    proof_facts: invoke_tx.proof_facts.clone(),
+                    proof: invoke_tx.proof.clone(),
+                    verification_task,
+                })
+            } else {
+                None
+            }
+        } else {
+            None
+        };
+
     mock_transaction_converter
         .expect_convert_rpc_tx_to_internal_rpc_tx()
         .once()
         .with(eq(rpc_tx))
-        .return_once(move |_| Ok(internal_tx));
+        .return_once(move |_| Ok((internal_tx, verification_handle)));
 
     let internal_tx = tx_args.get_internal_tx();
     let executable_tx = tx_args.get_executable_tx();
@@ -461,12 +481,15 @@ async fn test_add_tx_positive(
     Ok(executable_invoke_tx(invoke_args())),
 )]
 #[case::internal_to_executable_fails(
-    Ok(invoke_args().get_internal_tx()),
+    Ok((invoke_args().get_internal_tx(), None)),
     Err(TransactionConverterError::ClassNotFound { class_hash: ClassHash::default() })
 )]
 #[tokio::test]
 async fn test_transaction_converter_error(
-    #[case] expect_internal_rpc_tx_result: TransactionConverterResult<InternalRpcTransaction>,
+    #[case] expect_internal_rpc_tx_result: TransactionConverterResult<(
+        InternalRpcTransaction,
+        Option<VerificationHandle>,
+    )>,
     #[case] expect_executable_tx_result: TransactionConverterResult<AccountTransaction>,
     mut mock_dependencies: MockDependencies,
 ) {

crates/apollo_transaction_converter/src/lib.rs

@@ -9,4 +9,5 @@ pub use transaction_converter::{
     TransactionConverterError,
     TransactionConverterResult,
     TransactionConverterTrait,
+    VerificationHandle,
 };