feat(l1): admin_addPeer endpoint (#5004)

**Motivation**

Include geth's
[admin_addPeer](https://geth.ethereum.org/docs/interacting-with-geth/rpc/ns-admin#admin-addpeer)
endpoint since we found it necessary for running eip7934 hive tests with
the `ethereum/eels/consume-sync ` simulation.
Besides Nethermind, Reth, Besu, Erigon and other execution clients
implement it.

**Description**

This pr introduces the RPC endpoint `admin_addPeer`. Since this endpoint
needed to access p2p's generic server used to establish connections, the
`init_l1` function was restructure so the genserver is saved within the
peer_handler struct and can later be accessed from the `RPCApiContext`.

In `admin_addPeer` the connection is tried in a period of 10 seconds
waiting 100 milliseconds in between each message casted. This is because
connections are asynchronous so to check if the connection with the peer
was made we need to wait until it's established.

---------

Co-authored-by: SDartayet <[email protected]>
Co-authored-by: SDartayet <[email protected]>
Co-authored-by: ElFantasma <[email protected]>

Author: 3 people

Cargo.lock

@@ -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/Cargo.toml

@@ -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/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 {

cmd/ethrex/l2/initializers.rs

@@ -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/Cargo.toml

@@ -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/network.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(