feat: initial libp2p integration with req-resp status decoding

Cargo.toml

@@ -23,11 +23,21 @@ rust-version = "1.92.0"
 version = "0.1.0"
 
 [workspace.dependencies]
-ethlambda_types = { path = "crates/common/types" }
+ethlambda-fork-choice = { path = "crates/blockchain/fork_choice" }
+ethlambda-state-transition = { path = "crates/blockchain/state_transition" }
+ethlambda-crypto = { path = "crates/common/crypto" }
+ethlambda-types = { path = "crates/common/types" }
+ethlambda-p2p = { path = "crates/net/p2p" }
+ethlambda-rpc = { path = "crates/net/rpc" }
+ethlambda-storage = { path = "crates/storage" }
 
+tracing = "0.1"
+thiserror = "2.0.9"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1.0.117"
 
+tokio = "1.0"
+
 clap = { version = "4.3", features = ["derive", "env"] }
 
 # H256 type

Makefile

@@ -12,4 +12,4 @@ lint: ## 🔍 Run clippy on all workspace crates
 run-devnet: lean-quickstart ## 🚀 Run a local devnet using lean-quickstart
 	cargo build \
 	&& cd lean-quickstart \
-	&& NETWORK_DIR=local-devnet ./spin-node.sh --node ethlambda_0 --generateGenesis
+	&& NETWORK_DIR=local-devnet ./spin-node.sh --node zeam_0,ethlambda_0 --generateGenesis

bin/ethlambda/Cargo.toml

@@ -5,7 +5,13 @@ edition.workspace = true
 authors.workspace = true
 
 [dependencies]
-ethlambda_types.workspace = true
+ethlambda-p2p.workspace = true
+ethlambda-types.workspace = true
+
+tokio.workspace = true
+
+tracing.workspace = true
+tracing-subscriber = "0.3"
 
 serde.workspace = true
 ethereum-types.workspace = true

bin/ethlambda/src/main.rs

@@ -1,5 +1,8 @@
 use clap::Parser;
+use ethlambda_p2p::{parse_validators_file, start_p2p};
 use ethlambda_types::{genesis::Genesis, state::State};
+use tracing::info;
+use tracing_subscriber::{Registry, layer::SubscriberExt};
 
 const ASCII_ART: &str = r#"
       _   _     _                 _         _
@@ -13,9 +16,16 @@ const ASCII_ART: &str = r#"
 struct CliOptions {
     #[arg(long)]
     custom_genesis_json_file: String,
+    #[arg(long)]
+    validators_file: String,
+    #[arg(long)]
+    gossipsub_port: u16,
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
+    let subscriber = Registry::default().with(tracing_subscriber::fmt::layer());
+    tracing::subscriber::set_global_default(subscriber).unwrap();
     let options = CliOptions::parse();
 
     println!("{ASCII_ART}");
@@ -27,5 +37,19 @@ fn main() {
 
     let initial_state = State::from_genesis(&genesis);
 
+    let bootnodes = parse_validators_file(&options.validators_file);
+
+    let p2p_handle = tokio::spawn(start_p2p(bootnodes, options.gossipsub_port));
+
+    info!("Node initialized");
+
+    tokio::select! {
+        _ = p2p_handle => {
+            panic!("P2P node task has exited unexpectedly");
+        }
+        _ = tokio::signal::ctrl_c() => {
+            // Ctrl-C received, shutting down
+        }
+    }
     println!("Shutting down...");
 }

crates/blockchain/fork_choice/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ethlambda_fork_choice"
+name = "ethlambda-fork-choice"
 authors.workspace = true
 edition.workspace = true
 keywords.workspace = true

crates/blockchain/state_transition/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ethlambda_state_transition"
+name = "ethlambda-state-transition"
 authors.workspace = true
 edition.workspace = true
 keywords.workspace = true

crates/common/crypto/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ethlambda_crypto"
+name = "ethlambda-crypto"
 authors.workspace = true
 edition.workspace = true
 keywords.workspace = true

crates/common/types/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ethlambda_types"
+name = "ethlambda-types"
 authors.workspace = true
 edition.workspace = true
 keywords.workspace = true
@@ -10,5 +10,6 @@ rust-version.workspace = true
 version.workspace = true
 
 [dependencies]
+thiserror.workspace = true
 serde.workspace = true
 ethereum-types.workspace = true

crates/common/types/src/state.rs

@@ -52,13 +52,14 @@ impl State {
 
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 pub struct Checkpoint {
-    root: H256,
+    pub root: H256,
     // Used U256 due to it being serialized as string
-    slot: U256,
+    // TODO: use u64 and implement custom serialization
+    pub slot: U256,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct NetworkConfig {
-    num_validators: u64,
-    genesis_time: u64,
+    pub num_validators: u64,
+    pub genesis_time: u64,
 }

crates/net/p2p/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ethlambda_p2p"
+name = "ethlambda-p2p"
 authors.workspace = true
 edition.workspace = true
 keywords.workspace = true
@@ -10,3 +10,20 @@ rust-version.workspace = true
 version.workspace = true
 
 [dependencies]
+ethlambda-types.workspace = true
+
+async-trait = "0.1"
+
+libp2p = { version = "0.56", features = ["full"] }
+
+snap = "1.1"
+
+tracing.workspace = true
+
+# Required for NodeEnr parsing
+ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", rev = "1af63a4de7c93eb7413b9b003df1be82e1484c69" }
+ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", rev = "1af63a4de7c93eb7413b9b003df1be82e1484c69" }
+ethrex-common = { git = "https://github.com/lambdaclass/ethrex", rev = "1af63a4de7c93eb7413b9b003df1be82e1484c69" }
+
+[dev-dependencies]
+hex = "0.4"

crates/net/p2p/src/lib.rs

@@ -0,0 +1,178 @@
+use std::{net::IpAddr, time::Duration};
+
+use ethrex_common::H264;
+use ethrex_p2p::types::NodeRecord;
+use ethrex_rlp::decode::RLPDecode;
+use libp2p::{
+    Multiaddr, PeerId, StreamProtocol,
+    futures::StreamExt,
+    gossipsub::{self, MessageAuthenticity, ValidationMode},
+    identity::{PublicKey, secp256k1},
+    multiaddr::Protocol,
+    request_response::{self, Event, Message},
+    swarm::{NetworkBehaviour, SwarmEvent},
+};
+use tracing::{info, trace};
+
+use crate::messages::status::STATUS_PROTOCOL_V1;
+
+mod messages;
+
+pub async fn start_p2p(bootnodes: Vec<Bootnode>, listening_port: u16) {
+    let config = libp2p::gossipsub::ConfigBuilder::default()
+        // d
+        .mesh_n(8)
+        // d_low
+        .mesh_n_low(6)
+        // d_high
+        .mesh_n_high(12)
+        // d_lazy
+        .gossip_lazy(6)
+        .heartbeat_interval(Duration::from_millis(700))
+        .fanout_ttl(Duration::from_secs(60))
+        .history_length(6)
+        .history_gossip(3)
+        // seen_ttl_secs = seconds_per_slot * justification_lookback_slots * 2
+        .duplicate_cache_time(Duration::from_secs(4 * 3 * 2))
+        .validation_mode(ValidationMode::Anonymous)
+        .build()
+        .expect("invalid gossipsub config");
+
+    // TODO: setup custom message ID function
+    let gossipsub = libp2p::gossipsub::Behaviour::new(MessageAuthenticity::Anonymous, config)
+        .expect("failed to initiate behaviour");
+    let req_resp = request_response::Behaviour::new(
+        vec![(
+            StreamProtocol::new(STATUS_PROTOCOL_V1),
+            request_response::ProtocolSupport::Full,
+        )],
+        Default::default(),
+    );
+
+    let behavior = Behaviour {
+        gossipsub,
+        req_resp,
+    };
+
+    // TODO: set peer scoring params
+
+    // TODO: load identity from config or flag
+    let secret_key = secp256k1::SecretKey::try_from_bytes(
+        b")\x95PR\x9ay\xbc-\xce\x007G\xc5/\xb0c\x94e\xc8\x93\xe0\x0b\x04@\xacf\x14Mb^\x06j"
+            .to_vec(),
+    )
+    .unwrap();
+    let identity = libp2p::identity::Keypair::from(secp256k1::Keypair::from(secret_key));
+
+    // TODO: implement Executor with spawned?
+    // libp2p::swarm::Config::with_executor(executor)
+    let mut swarm = libp2p::SwarmBuilder::with_existing_identity(identity)
+        .with_tokio()
+        .with_quic()
+        .with_behaviour(|_| behavior)
+        .expect("failed to add behaviour to swarm")
+        .with_swarm_config(|config| {
+            // Disable idle connection timeout
+            config.with_idle_connection_timeout(Duration::from_secs(u64::MAX))
+        })
+        .build();
+    for bootnode in bootnodes {
+        let addr = Multiaddr::empty()
+            .with(bootnode.ip.into())
+            .with(Protocol::Udp(bootnode.quic_port))
+            .with(Protocol::QuicV1)
+            .with_p2p(PeerId::from_public_key(&bootnode.public_key))
+            .expect("failed to add peer ID to multiaddr");
+        swarm.dial(addr).unwrap();
+    }
+    let addr = Multiaddr::empty()
+        .with("127.0.0.1".parse::<IpAddr>().unwrap().into())
+        .with(Protocol::Udp(listening_port))
+        .with(Protocol::QuicV1);
+    swarm
+        .listen_on(addr)
+        .expect("failed to bind gossipsub listening address");
+
+    println!("P2P node started on port {listening_port}");
+
+    event_loop(swarm).await;
+}
+
+/// [libp2p Behaviour](libp2p::swarm::NetworkBehaviour) combining Gossipsub and Request-Response Behaviours
+#[derive(NetworkBehaviour)]
+struct Behaviour {
+    gossipsub: gossipsub::Behaviour,
+    req_resp: request_response::Behaviour<messages::status::StatusCodec>,
+}
+
+/// Event loop for the P2P crate.
+/// Processes swarm events, including incoming requests, responses, and gossip.
+async fn event_loop(mut swarm: libp2p::Swarm<Behaviour>) {
+    while let Some(event) = swarm.next().await {
+        match event {
+            SwarmEvent::Behaviour(BehaviourEvent::ReqResp(Event::Message {
+                peer,
+                connection_id,
+                message:
+                    Message::Request {
+                        request_id,
+                        request,
+                        channel,
+                    },
+            })) => {
+                info!(finalized_slot=%request.finalized.slot, head_slot=%request.head.slot, "Received status request from peer {peer}");
+            }
+            _ => {
+                trace!(?event, "Ignored swarm event");
+            }
+        }
+    }
+}
+
+pub struct Bootnode {
+    ip: IpAddr,
+    quic_port: u16,
+    public_key: PublicKey,
+}
+
+pub fn parse_validators_file(bootnodes_path: &str) -> Vec<Bootnode> {
+    let bootnodes_yaml =
+        std::fs::read_to_string(bootnodes_path).expect("Failed to read validators.yaml");
+
+    let mut bootnodes = vec![];
+
+    // File is YAML, but we try to avoid pulling a full YAML parser just for this
+    for line in bootnodes_yaml.lines() {
+        let trimmed_line = line.trim();
+        if trimmed_line.is_empty() {
+            continue;
+        }
+        let enr_str = trimmed_line.strip_prefix("- ").unwrap();
+        let base64_decoded = ethrex_common::base64::decode(&enr_str.as_bytes()[4..]);
+        let record = NodeRecord::decode(&base64_decoded).unwrap();
+        let (_, quic_port_bytes) = record
+            .pairs
+            .iter()
+            .find(|(key, _)| key.as_ref() == b"quic")
+            .expect("node doesn't support QUIC");
+
+        let (_, public_key_rlp) = record
+            .pairs
+            .iter()
+            .find(|(key, _)| key.as_ref() == b"secp256k1")
+            .expect("node record missing public key");
+
+        let public_key_bytes = H264::decode(public_key_rlp).unwrap();
+        let public_key =
+            libp2p::identity::secp256k1::PublicKey::try_from_bytes(public_key_bytes.as_bytes())
+                .unwrap();
+
+        let quic_port = u16::decode(quic_port_bytes.as_ref()).unwrap();
+        bootnodes.push(Bootnode {
+            ip: "127.0.0.1".parse().unwrap(),
+            quic_port,
+            public_key: public_key.into(),
+        });
+    }
+    bootnodes
+}

crates/net/p2p/src/messages/mod.rs

@@ -0,0 +1 @@
+pub mod status;