feat(l1): implement missing p2p message handlers (#1136)

**Motivation**
`p2p` messages `Disconnect`, `Ping` and `Pong` handlers where not
implemented

**Description**

- Improved `Disconnect` message handler
- Now we send `Ping` messages every 15 seconds (same as go-ethereum)
- We ignore `Pong` messages (same as go-ethereum)

Closes #1030 

**New Related issues**
#1134 
#1135

crates/networking/p2p/Cargo.toml

@@ -16,6 +16,7 @@ bytes.workspace = true
 hex.workspace = true
 thiserror.workspace = true
 lazy_static.workspace = true
+snap.workspace = true
 
 k256 = { version = "0.13.3", features = ["ecdh"] }
 sha3 = "0.10.8"
@@ -26,7 +27,6 @@ hmac = "0.12.1"
 aes = "0.8.4"
 ctr = "0.9.2"
 rand = "0.8.5"
-snap = "1.1.1"
 
 [dev-dependencies]
 hex-literal = "0.4.1"

crates/networking/p2p/net.rs

@@ -18,7 +18,7 @@ use k256::{
 };
 use kademlia::{bucket_number, KademliaTable, MAX_NODES_PER_BUCKET};
 use rand::rngs::OsRng;
-use rlpx::{connection::RLPxConnection, message::Message as RLPxMessage};
+use rlpx::{connection::RLPxConnection, error::RLPxError, message::Message as RLPxMessage};
 use tokio::{
     net::{TcpSocket, TcpStream, UdpSocket},
     sync::{broadcast, Mutex},
@@ -809,23 +809,32 @@ async fn handle_peer_as_initiator(
 }
 
 async fn handle_peer(mut conn: RLPxConnection<TcpStream>, table: Arc<Mutex<KademliaTable>>) {
-    match conn.handshake().await {
-        Ok(_) => match conn.handle_peer().await {
-            Ok(_) => unreachable!(),
-            Err(e) => info!("Error during RLPx connection: ({e})"),
-        },
-        Err(e) => {
-            if let Ok(node_id) = conn.get_remote_node_id() {
-                // Discard peer from kademlia table
-                info!("Handshake failed: ({e}), discarding peer {node_id}");
-                table.lock().await.replace_peer(node_id);
-            } else {
-                info!("Handshake failed: ({e}), unknown peer");
-            }
+    // Perform handshake
+    if let Err(e) = conn.handshake().await {
+        peer_conn_failed("Handshake failed", e, conn, table).await;
+    } else {
+        // Handshake OK: handle connection
+        if let Err(e) = conn.handle_peer_conn().await {
+            peer_conn_failed("Error during RLPx connection", e, conn, table).await;
         }
     }
 }
 
+async fn peer_conn_failed(
+    error_text: &str,
+    error: RLPxError,
+    conn: RLPxConnection<TcpStream>,
+    table: Arc<Mutex<KademliaTable>>,
+) {
+    if let Ok(node_id) = conn.get_remote_node_id() {
+        // Discard peer from kademlia table
+        info!("{error_text}: ({error}), discarding peer {node_id}");
+        table.lock().await.replace_peer(node_id);
+    } else {
+        info!("{error_text}: ({error}), unknown peer")
+    }
+}
+
 pub fn node_id_from_signing_key(signer: &SigningKey) -> H512 {
     let public_key = PublicKey::from(signer.verifying_key());
     let encoded = public_key.to_encoded_point(false);