feat(mempool): add basic p2p agent interface with unimplemented code

Cargo.toml

@@ -14,6 +14,7 @@ members = [
   "crates/mempool",
   "crates/mempool_infra",
   "crates/mempool_node",
+  "crates/mempool_p2p",
   "crates/mempool_test_utils",
   "crates/mempool_types",
   "crates/native_blockifier",

crates/mempool_p2p/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "starknet_mempool_p2p"
+version = "0.0.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+async-trait.workspace = true
+papyrus_network.workspace = true
+papyrus_proc_macros.workspace = true
+serde = { workspace = true, features = ["derive"] }
+starknet_api.workspace = true
+starknet_mempool_infra.workspace = true
+thiserror.workspace = true

crates/mempool_p2p/src/lib.rs

@@ -0,0 +1,13 @@
+pub mod receiver;
+pub mod sender;
+
+use papyrus_network::NetworkConfig;
+
+use crate::receiver::MempoolP2pReceiver;
+use crate::sender::MempoolP2pSender;
+
+pub fn create_p2p_sender_and_receiver(
+    _network_config: NetworkConfig,
+) -> (MempoolP2pSender, MempoolP2pReceiver) {
+    unimplemented!()
+}

crates/mempool_p2p/src/receiver/mod.rs

@@ -0,0 +1,11 @@
+use async_trait::async_trait;
+use starknet_mempool_infra::component_runner::{ComponentStartError, ComponentStarter};
+
+pub struct MempoolP2pReceiver;
+
+#[async_trait]
+impl ComponentStarter for MempoolP2pReceiver {
+    async fn start(&mut self) -> Result<(), ComponentStartError> {
+        unimplemented!()
+    }
+}

crates/mempool_p2p/src/sender/mod.rs

@@ -0,0 +1,104 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+pub use papyrus_network::network_manager::BroadcastedMessageManager;
+use papyrus_proc_macros::handle_response_variants;
+use serde::{Deserialize, Serialize};
+use starknet_api::rpc_transaction::RpcTransaction;
+use starknet_mempool_infra::component_client::{ClientError, LocalComponentClient};
+use starknet_mempool_infra::component_definitions::ComponentRequestHandler;
+use thiserror::Error;
+
+pub struct MempoolP2pSender;
+
+// This error is defined even though it's empty to be compatible with the other components.
+#[derive(Debug, Error, Serialize, Deserialize, Clone)]
+pub enum MempoolP2pSenderError {}
+
+pub type MempoolP2pSenderResult<T> = Result<T, MempoolP2pSenderError>;
+
+#[derive(Clone, Debug, Error)]
+pub enum MempoolP2pSenderClientError {
+    #[error(transparent)]
+    ClientError(#[from] ClientError),
+    #[error(transparent)]
+    MempoolP2pSenderError(#[from] MempoolP2pSenderError),
+}
+
+pub type MempoolP2pSenderClientResult<T> = Result<T, MempoolP2pSenderClientError>;
+
+#[async_trait]
+pub trait MempoolP2pSenderClient: Send + Sync {
+    /// Adds a transaction to be propagated to other peers. This should only be called on a new
+    /// transaction coming from the user and not from another peer. To handle transactions coming
+    /// from other peers, use `continue_propagation`.
+    async fn add_transaction(
+        &self,
+        transaction: RpcTransaction,
+    ) -> MempoolP2pSenderClientResult<()>;
+
+    /// Continues the propagation of a transaction we've received from another peer.
+    async fn continue_propagation(
+        &self,
+        propagation_manager: BroadcastedMessageManager,
+    ) -> MempoolP2pSenderClientResult<()>;
+}
+
+pub type SharedMempoolP2pSenderClient = Arc<dyn MempoolP2pSenderClient>;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum MempoolP2pSenderRequest {
+    AddTransaction(RpcTransaction),
+    ContinuePropagation(BroadcastedMessageManager),
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum MempoolP2pSenderResponse {
+    AddTransaction(MempoolP2pSenderResult<()>),
+    ContinuePropagation(MempoolP2pSenderResult<()>),
+}
+
+#[async_trait]
+impl ComponentRequestHandler<MempoolP2pSenderRequest, MempoolP2pSenderResponse>
+    for MempoolP2pSender
+{
+    async fn handle_request(
+        &mut self,
+        _request: MempoolP2pSenderRequest,
+    ) -> MempoolP2pSenderResponse {
+        unimplemented!()
+    }
+}
+
+#[async_trait]
+impl MempoolP2pSenderClient
+    for LocalComponentClient<MempoolP2pSenderRequest, MempoolP2pSenderResponse>
+{
+    async fn add_transaction(
+        &self,
+        transaction: RpcTransaction,
+    ) -> MempoolP2pSenderClientResult<()> {
+        let request = MempoolP2pSenderRequest::AddTransaction(transaction);
+        let response = self.send(request).await;
+        handle_response_variants!(
+            MempoolP2pSenderResponse,
+            AddTransaction,
+            MempoolP2pSenderClientError,
+            MempoolP2pSenderError
+        )
+    }
+
+    async fn continue_propagation(
+        &self,
+        propagation_manager: BroadcastedMessageManager,
+    ) -> MempoolP2pSenderClientResult<()> {
+        let request = MempoolP2pSenderRequest::ContinuePropagation(propagation_manager);
+        let response = self.send(request).await;
+        handle_response_variants!(
+            MempoolP2pSenderResponse,
+            ContinuePropagation,
+            MempoolP2pSenderClientError,
+            MempoolP2pSenderError
+        )
+    }
+}

crates/mempool_types/src/communication.rs

@@ -30,6 +30,8 @@ pub type SharedMempoolClient = Arc<dyn MempoolClient>;
 #[automock]
 #[async_trait]
 pub trait MempoolClient: Send + Sync {
+    // TODO: Add Option<BroadcastedMessageManager> as an argument for add_transaction
+    // TODO: Rename tx to transaction
     async fn add_tx(&self, mempool_input: MempoolInput) -> MempoolClientResult<()>;
     async fn get_txs(&self, n_txs: usize) -> MempoolClientResult<Vec<Transaction>>;
 }

crates/papyrus_network/src/network_manager/mod.rs

@@ -21,6 +21,7 @@ use libp2p::swarm::SwarmEvent;
 use libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm};
 use metrics::gauge;
 use papyrus_common::metrics as papyrus_metrics;
+use serde::{Deserialize, Serialize};
 use sqmr::Bytes;
 use tracing::{debug, error, info, trace, warn};
 
@@ -838,7 +839,7 @@ pub type BroadcastTopicSender<T> = With<
 >;
 
 // TODO(alonl): remove clone
-#[derive(Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BroadcastedMessageManager {
     peer_id: PeerId,
 }