feat(l1): admin_addPeer endpoint

cmd/ethrex/Cargo.toml

@@ -52,6 +52,9 @@ thiserror.workspace = true
 itertools = "0.14.0"
 url.workspace = true
 
+spawned-rt.workspace = true
+spawned-concurrency.workspace = true
+
 # L2 external dependencies
 
 tui-logger = { workspace = true, optional = true }

cmd/ethrex/initializers.rs

@@ -11,8 +11,7 @@ use ethrex_common::types::Genesis;
 use ethrex_config::networks::Network;
 
 use ethrex_metrics::profiling::{FunctionProfilingLayer, initialize_block_processing_profile};
-#[cfg(feature = "l2")]
-use ethrex_p2p::rlpx::l2::l2_connection::P2PBasedContext;
+use ethrex_p2p::rlpx::initiator::RLPxInitiator;
 use ethrex_p2p::{
     discv4::peer_table::PeerTable,
     network::P2PContext,
@@ -193,13 +192,10 @@ pub async fn init_network(
     opts: &Options,
     network: &Network,
     datadir: &Path,
-    local_p2p_node: Node,
-    signer: SecretKey,
     peer_handler: PeerHandler,
-    store: Store,
     tracker: TaskTracker,
     blockchain: Arc<Blockchain>,
-    #[cfg(feature = "l2")] based_context: Option<P2PBasedContext>,
+    context: P2PContext,
 ) {
     if opts.dev {
         error!("Binary wasn't built with The feature flag `dev` enabled.");
@@ -210,26 +206,6 @@ pub async fn init_network(
 
     let bootnodes = get_bootnodes(opts, network, datadir);
 
-    #[cfg(feature = "l2")]
-    let based_context_arg = based_context;
-
-    #[cfg(not(feature = "l2"))]
-    let based_context_arg = None;
-
-    let context = P2PContext::new(
-        local_p2p_node,
-        tracker.clone(),
-        signer,
-        peer_handler.peer_table.clone(),
-        store,
-        blockchain.clone(),
-        get_client_version(),
-        based_context_arg,
-        opts.tx_broadcasting_time_interval,
-    )
-    .await
-    .expect("P2P context could not be created");
-
     ethrex_p2p::start_network(context, bootnodes)
         .await
         .expect("Network starts");
@@ -439,17 +415,35 @@ pub async fn init_l1(
 
     let local_node_record = get_local_node_record(datadir, &local_p2p_node, &signer);
 
-    let peer_handler = PeerHandler::new(PeerTable::spawn(opts.target_peers));
+    let peer_table = PeerTable::spawn(opts.target_peers);
 
     // TODO: Check every module starts properly.
     let tracker = TaskTracker::new();
 
     let cancel_token = tokio_util::sync::CancellationToken::new();
 
+    let p2p_context = P2PContext::new(
+        local_p2p_node.clone(),
+        tracker.clone(),
+        signer,
+        peer_table.clone(),
+        store.clone(),
+        blockchain.clone(),
+        get_client_version(),
+        None,
+        opts.tx_broadcasting_time_interval,
+    )
+    .await
+    .expect("P2P context could not be created");
+
+    let initiator = RLPxInitiator::spawn(p2p_context.clone()).await;
+
+    let peer_handler = PeerHandler::new(peer_table.clone(), initiator);
+
     init_rpc_api(
         &opts,
         peer_handler.clone(),
-        local_p2p_node.clone(),
+        local_p2p_node,
         local_node_record.clone(),
         store.clone(),
         blockchain.clone(),
@@ -471,14 +465,10 @@ pub async fn init_l1(
             &opts,
             &network,
             datadir,
-            local_p2p_node,
-            signer,
             peer_handler.clone(),
-            store.clone(),
             tracker.clone(),
             blockchain.clone(),
-            #[cfg(feature = "l2")]
-            None,
+            p2p_context,
         )
         .await;
     } else {

cmd/ethrex/l2/initializers.rs

@@ -15,8 +15,9 @@ use ethrex_l2::SequencerConfig;
 use ethrex_l2::sequencer::l1_committer::regenerate_head_state;
 use ethrex_p2p::{
     discv4::peer_table::PeerTable,
+    network::P2PContext,
     peer_handler::PeerHandler,
-    rlpx::l2::l2_connection::P2PBasedContext,
+    rlpx::{initiator::RLPxInitiator, l2::l2_connection::P2PBasedContext},
     sync_manager::SyncManager,
     types::{Node, NodeRecord},
 };
@@ -35,7 +36,7 @@ use url::Url;
 async fn init_rpc_api(
     opts: &L1Options,
     l2_opts: &L2Options,
-    peer_table: PeerTable,
+    peer_handler: PeerHandler,
     local_p2p_node: Node,
     local_node_record: NodeRecord,
     store: Store,
@@ -46,8 +47,6 @@ async fn init_rpc_api(
     log_filter_handler: Option<reload::Handle<EnvFilter, Registry>>,
     gas_ceil: Option<u64>,
 ) {
-    let peer_handler = PeerHandler::new(peer_table);
-
     init_datadir(&opts.datadir);
 
     // Create SyncManager
@@ -198,18 +197,51 @@ pub async fn init_l2(
 
     let local_node_record = get_local_node_record(&datadir, &local_p2p_node, &signer);
 
-    let peer_handler = PeerHandler::new(PeerTable::spawn(opts.node_opts.target_peers));
+    let peer_table = PeerTable::spawn(opts.node_opts.target_peers);
 
     // TODO: Check every module starts properly.
     let tracker = TaskTracker::new();
     let mut join_set = JoinSet::new();
 
+    let p2p_context = P2PContext::new(
+        local_p2p_node.clone(),
+        tracker.clone(),
+        signer,
+        peer_table.clone(),
+        store.clone(),
+        blockchain.clone(),
+        get_client_version(),
+        #[cfg(feature = "l2")]
+        Some(P2PBasedContext {
+            store_rollup: rollup_store.clone(),
+            // TODO: The Web3Signer refactor introduced a limitation where the committer key cannot be accessed directly because the signer could be either Local or Remote.
+            // The Signer enum cannot be used in the P2PBasedContext struct due to cyclic dependencies between the l2-rpc and p2p crates.
+            // As a temporary solution, a dummy committer key is used until a proper mechanism to utilize the Signer enum is implemented.
+            // This should be replaced with the Signer enum once the refactor is complete.
+            committer_key: Arc::new(
+                SecretKey::from_slice(
+                    &hex::decode(
+                        "385c546456b6a603a1cfcaa9ec9494ba4832da08dd6bcf4de9a71e4a01b74924",
+                    )
+                    .expect("Invalid committer key"),
+                )
+                .expect("Failed to create committer key"),
+            ),
+        }),
+        opts.node_opts.tx_broadcasting_time_interval,
+    )
+    .await
+    .expect("P2P context could not be created");
+
+    let initiator = RLPxInitiator::spawn(p2p_context.clone()).await;
+    let peer_handler = PeerHandler::new(PeerTable::spawn(opts.node_opts.target_peers), initiator);
+
     let cancel_token = tokio_util::sync::CancellationToken::new();
 
     init_rpc_api(
         &opts.node_opts,
         &opts,
-        peer_handler.peer_table.clone(),
+        peer_handler.clone(),
         local_p2p_node.clone(),
         local_node_record.clone(),
         store.clone(),
@@ -241,28 +273,10 @@ pub async fn init_l2(
             &opts.node_opts,
             &network,
             &datadir,
-            local_p2p_node,
-            signer,
             peer_handler.clone(),
-            store.clone(),
             tracker,
             blockchain.clone(),
-            Some(P2PBasedContext {
-                store_rollup: rollup_store.clone(),
-                // TODO: The Web3Signer refactor introduced a limitation where the committer key cannot be accessed directly because the signer could be either Local or Remote.
-                // The Signer enum cannot be used in the P2PBasedContext struct due to cyclic dependencies between the l2-rpc and p2p crates.
-                // As a temporary solution, a dummy committer key is used until a proper mechanism to utilize the Signer enum is implemented.
-                // This should be replaced with the Signer enum once the refactor is complete.
-                committer_key: Arc::new(
-                    SecretKey::from_slice(
-                        &hex::decode(
-                            "385c546456b6a603a1cfcaa9ec9494ba4832da08dd6bcf4de9a71e4a01b74924",
-                        )
-                        .expect("Invalid committer key"),
-                    )
-                    .expect("Failed to create committer key"),
-                ),
-            }),
+            p2p_context,
         )
         .await;
     } else {

crates/networking/p2p/Cargo.toml

@@ -62,6 +62,7 @@ default = ["c-kzg"]
 c-kzg = ["ethrex-blockchain/c-kzg", "ethrex-common/c-kzg"]
 sync-test = []
 l2 = ["dep:ethrex-storage-rollup"]
+test-utils = []
 
 [lints.clippy]
 unwrap_used = "deny"

crates/networking/p2p/network.rs

@@ -11,7 +11,6 @@ use crate::{
     metrics::METRICS,
     rlpx::{
         connection::server::{PeerConnBroadcastSender, PeerConnection},
-        initiator::RLPxInitiator,
         message::Message,
         p2p::SUPPORTED_SNAP_CAPABILITIES,
     },
@@ -94,6 +93,36 @@ impl P2PContext {
             tx_broadcaster,
         })
     }
+
+    #[cfg(any(test, feature = "test-utils"))]
+    /// Creates a dummy P2PContext for tests
+    /// This should only be used in tests as it won't be able to connect to the p2p network
+    pub async fn dummy(peer_table: PeerTable) -> P2PContext {
+        use ethrex_storage::EngineType;
+
+        let storage = Store::new("./temp", EngineType::InMemory).expect("Failed to create Store");
+        let blockchain: Arc<Blockchain> = Arc::new(Blockchain::default_with_store(storage.clone()));
+        let local_node = Node::from_enode_url(
+            "enode://d860a01f9722d78051619d1e2351aba3f43f943f6f00718d1b9baa4101932a1f5011f16bb2b1bb35db20d6fe28fa0bf09636d26a87d31de9ec6203eeedb1f666@18.138.108.67:30303",
+        ).expect("Bad enode url");
+        let (channel_broadcast_send_end, _) =
+            tokio::sync::broadcast::channel::<(tokio::task::Id, Arc<Message>)>(100000);
+        P2PContext {
+            tracker: TaskTracker::default(),
+            signer: SecretKey::from_byte_array(&[0xcd; 32]).expect("32 bytes, within curve order"),
+            table: peer_table.clone(),
+            storage,
+            blockchain: blockchain.clone(),
+            broadcast: channel_broadcast_send_end,
+            local_node: local_node.clone(),
+            client_version: "".to_string(),
+            #[cfg(feature = "l2")]
+            based_context: None,
+            tx_broadcaster: TxBroadcaster::spawn(peer_table.clone(), blockchain, 1000)
+                .await
+                .expect("Failed to spawn tx broadcaster"),
+        }
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -123,8 +152,6 @@ pub async fn start_network(context: P2PContext, bootnodes: Vec<Node>) -> Result<
         error!("Failed to start discovery server: {e}");
     })?;
 
-    RLPxInitiator::spawn(context.clone()).await;
-
     context.tracker.spawn(serve_p2p_requests(context.clone()));
 
     Ok(())

crates/networking/p2p/peer_handler.rs

@@ -1,5 +1,8 @@
+#[cfg(any(test, feature = "test-utils"))]
+use crate::discv4::peer_table::TARGET_PEERS;
+use crate::rlpx::initiator::RLPxInitiator;
 use crate::{
-    discv4::peer_table::{PeerData, PeerTable, PeerTableError, TARGET_PEERS},
+    discv4::peer_table::{PeerData, PeerTable, PeerTableError},
     metrics::{CurrentStepValue, METRICS},
     rlpx::{
         connection::server::PeerConnection,
@@ -34,6 +37,7 @@ use ethrex_rlp::{decode::RLPDecode, encode::RLPEncode};
 use ethrex_storage::Store;
 use ethrex_trie::Nibbles;
 use ethrex_trie::{Node, verify_range};
+use spawned_concurrency::tasks::GenServerHandle;
 use std::{
     collections::{BTreeMap, HashMap, HashSet, VecDeque},
     io::ErrorKind,
@@ -68,6 +72,7 @@ pub const MAX_BLOCK_BODIES_TO_REQUEST: usize = 128;
 #[derive(Debug, Clone)]
 pub struct PeerHandler {
     pub peer_table: PeerTable,
+    pub initiator: GenServerHandle<RLPxInitiator>,
 }
 
 pub enum BlockRequestOrder {
@@ -142,14 +147,19 @@ async fn ask_peer_head_number(
 }
 
 impl PeerHandler {
-    pub fn new(peer_table: PeerTable) -> PeerHandler {
-        Self { peer_table }
+    pub fn new(peer_table: PeerTable, initiator: GenServerHandle<RLPxInitiator>) -> PeerHandler {
+        Self {
+            peer_table,
+            initiator,
+        }
     }
 
+    #[cfg(any(test, feature = "test-utils"))]
     /// Creates a dummy PeerHandler for tests where interacting with peers is not needed
     /// This should only be used in tests as it won't be able to interact with the node's connected peers
-    pub fn dummy() -> PeerHandler {
-        PeerHandler::new(PeerTable::spawn(TARGET_PEERS))
+    pub async fn dummy() -> PeerHandler {
+        let peer_table = PeerTable::spawn(TARGET_PEERS);
+        PeerHandler::new(peer_table.clone(), RLPxInitiator::dummy(peer_table).await)
     }
 
     async fn make_request(