feat(starknet_mempool_p2p): add class manager to propogator

Author: noamsp-starkware

Cargo.lock

@@ -16,6 +16,7 @@ papyrus_network.workspace = true
 papyrus_protobuf.workspace = true
 serde.workspace = true
 starknet_api.workspace = true
+starknet_class_manager_types.workspace = true
 starknet_gateway_types.workspace = true
 starknet_mempool_p2p_types.workspace = true
 starknet_sequencer_infra.workspace = true

crates/starknet_mempool_p2p/Cargo.toml

@@ -5,6 +5,8 @@ pub mod runner;
 use futures::FutureExt;
 use papyrus_network::gossipsub_impl::Topic;
 use papyrus_network::network_manager::{BroadcastTopicChannels, NetworkManager};
+use starknet_class_manager_types::transaction_converter::TransactionConverter;
+use starknet_class_manager_types::SharedClassManagerClient;
 use starknet_gateway_types::communication::SharedGatewayClient;
 
 use crate::config::MempoolP2pConfig;
@@ -16,7 +18,12 @@ pub const MEMPOOL_TOPIC: &str = "starknet_mempool_transaction_propagation/0.1.0"
 pub fn create_p2p_propagator_and_runner(
     mempool_p2p_config: MempoolP2pConfig,
     gateway_client: SharedGatewayClient,
+    class_manager_client: SharedClassManagerClient,
 ) -> (MempoolP2pPropagator, MempoolP2pRunner) {
+    let transaction_converter = TransactionConverter::new(
+        class_manager_client.clone(),
+        mempool_p2p_config.network_config.chain_id.clone(),
+    );
     let mut network_manager = NetworkManager::new(
         mempool_p2p_config.network_config,
         // TODO(Shahak): Consider filling this once the sequencer node has a name.
@@ -30,7 +37,8 @@ pub fn create_p2p_propagator_and_runner(
             )
             .expect("Failed to register broadcast topic");
     let network_future = network_manager.run();
-    let mempool_p2p_propagator = MempoolP2pPropagator::new(broadcast_topic_client.clone());
+    let mempool_p2p_propagator =
+        MempoolP2pPropagator::new(broadcast_topic_client.clone(), transaction_converter);
     let mempool_p2p_runner = MempoolP2pRunner::new(
         network_future.boxed(),
         broadcasted_messages_receiver,

crates/starknet_mempool_p2p/src/lib.rs

@@ -4,6 +4,7 @@ mod test;
 use async_trait::async_trait;
 use papyrus_network::network_manager::{BroadcastTopicClient, BroadcastTopicClientTrait};
 use papyrus_protobuf::mempool::RpcTransactionWrapper;
+use starknet_class_manager_types::transaction_converter::TransactionConverter;
 use starknet_mempool_p2p_types::communication::{
     MempoolP2pPropagatorRequest,
     MempoolP2pPropagatorResponse,
@@ -15,11 +16,15 @@ use tracing::warn;
 
 pub struct MempoolP2pPropagator {
     broadcast_topic_client: BroadcastTopicClient<RpcTransactionWrapper>,
+    _transaction_converter: TransactionConverter,
 }
 
 impl MempoolP2pPropagator {
-    pub fn new(broadcast_topic_client: BroadcastTopicClient<RpcTransactionWrapper>) -> Self {
-        Self { broadcast_topic_client }
+    pub fn new(
+        broadcast_topic_client: BroadcastTopicClient<RpcTransactionWrapper>,
+        transaction_converter: TransactionConverter,
+    ) -> Self {
+        Self { broadcast_topic_client, _transaction_converter: transaction_converter }
     }
 }
 

crates/starknet_mempool_p2p/src/propagator/mod.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use futures::stream::StreamExt;
 use papyrus_network::network_manager::test_utils::{
     mock_register_broadcast_topic,
@@ -8,7 +10,10 @@ use papyrus_network::network_manager::BroadcastTopicChannels;
 use papyrus_network_types::network_types::BroadcastedMessageMetadata;
 use papyrus_protobuf::mempool::RpcTransactionWrapper;
 use papyrus_test_utils::{get_rng, GetTestInstance};
+use starknet_api::core::ChainId;
 use starknet_api::rpc_transaction::RpcTransaction;
+use starknet_class_manager_types::transaction_converter::TransactionConverter;
+use starknet_class_manager_types::EmptyClassManagerClient;
 use starknet_mempool_p2p_types::communication::MempoolP2pPropagatorRequest;
 use starknet_sequencer_infra::component_definitions::ComponentRequestHandler;
 use tokio::time::timeout;
@@ -25,7 +30,11 @@ async fn process_handle_add_tx() {
         subscriber_channels;
     let BroadcastNetworkMock { mut messages_to_broadcast_receiver, .. } = mock_network;
     let rpc_transaction = RpcTransaction::get_test_instance(&mut get_rng());
-    let mut mempool_p2p_propagator = MempoolP2pPropagator::new(broadcast_topic_client);
+    // TODO(noamsp): use MockTransactionConverterTrait
+    let transaction_converter =
+        TransactionConverter::new(Arc::new(EmptyClassManagerClient), ChainId::create_for_testing());
+    let mut mempool_p2p_propagator =
+        MempoolP2pPropagator::new(broadcast_topic_client, transaction_converter);
     mempool_p2p_propagator
         .handle_request(MempoolP2pPropagatorRequest::AddTransaction(rpc_transaction.clone()))
         .await;
@@ -41,7 +50,11 @@ async fn process_handle_continue_propagation() {
         subscriber_channels;
     let BroadcastNetworkMock { mut continue_propagation_receiver, .. } = mock_network;
     let propagation_metadata = BroadcastedMessageMetadata::get_test_instance(&mut get_rng());
-    let mut mempool_p2p_propagator = MempoolP2pPropagator::new(broadcast_topic_client);
+    // TODO(noamsp): use MockTransactionConverterTrait
+    let transaction_converter =
+        TransactionConverter::new(Arc::new(EmptyClassManagerClient), ChainId::create_for_testing());
+    let mut mempool_p2p_propagator =
+        MempoolP2pPropagator::new(broadcast_topic_client, transaction_converter);
     mempool_p2p_propagator
         .handle_request(MempoolP2pPropagatorRequest::ContinuePropagation(
             propagation_metadata.clone(),

crates/starknet_mempool_p2p/src/propagator/test.rs

@@ -103,23 +103,26 @@ pub fn create_node_components(
         ActiveComponentExecutionMode::Disabled => None,
     };
 
-    let (mempool_p2p_propagator, mempool_p2p_runner) = match config
-        .components
-        .mempool_p2p
-        .execution_mode
-    {
-        ReactiveComponentExecutionMode::LocalExecutionWithRemoteDisabled
-        | ReactiveComponentExecutionMode::LocalExecutionWithRemoteEnabled => {
-            let gateway_client =
-                clients.get_gateway_shared_client().expect("Gateway Client should be available");
-            let (mempool_p2p_propagator, mempool_p2p_runner) =
-                create_p2p_propagator_and_runner(config.mempool_p2p_config.clone(), gateway_client);
-            (Some(mempool_p2p_propagator), Some(mempool_p2p_runner))
-        }
-        ReactiveComponentExecutionMode::Disabled | ReactiveComponentExecutionMode::Remote => {
-            (None, None)
-        }
-    };
+    let (mempool_p2p_propagator, mempool_p2p_runner) =
+        match config.components.mempool_p2p.execution_mode {
+            ReactiveComponentExecutionMode::LocalExecutionWithRemoteDisabled
+            | ReactiveComponentExecutionMode::LocalExecutionWithRemoteEnabled => {
+                let gateway_client = clients
+                    .get_gateway_shared_client()
+                    .expect("Gateway Client should be available");
+                // TODO: Remove this and use the real client instead once implemented.
+                let class_manager_client = Arc::new(EmptyClassManagerClient);
+                let (mempool_p2p_propagator, mempool_p2p_runner) = create_p2p_propagator_and_runner(
+                    config.mempool_p2p_config.clone(),
+                    gateway_client,
+                    class_manager_client,
+                );
+                (Some(mempool_p2p_propagator), Some(mempool_p2p_runner))
+            }
+            ReactiveComponentExecutionMode::Disabled | ReactiveComponentExecutionMode::Remote => {
+                (None, None)
+            }
+        };
 
     let mempool = match config.components.mempool.execution_mode {
         ReactiveComponentExecutionMode::LocalExecutionWithRemoteDisabled