From 64da9f007c4534d12e46bf050f0e095255766cf3 Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 1 May 2026 22:20:26 +0300 Subject: [PATCH 1/2] Move node name registry to P2PServer --- bin/ethlambda/src/main.rs | 13 +++---- crates/net/p2p/src/lib.rs | 64 ++++++++++++++++++++++++++++++----- crates/net/p2p/src/metrics.rs | 49 +++------------------------ 3 files changed, 68 insertions(+), 58 deletions(-) diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 3c3f816c..79e206f1 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -20,7 +20,7 @@ use std::{ use clap::Parser; use ethlambda_blockchain::key_manager::ValidatorKeyPair; use ethlambda_network_api::{InitBlockChain, InitP2P, ToBlockChainToP2PRef, ToP2PToBlockChainRef}; -use ethlambda_p2p::{Bootnode, P2P, SwarmConfig, build_swarm, parse_enrs}; +use ethlambda_p2p::{Bootnode, P2P, PeerId, SwarmConfig, build_swarm, parse_enrs}; use ethlambda_types::primitives::H256; use ethlambda_types::{ aggregator::AggregatorController, @@ -141,7 +141,7 @@ async fn main() -> eyre::Result<()> { "Loaded genesis configuration" ); - populate_name_registry(&validator_config); + let node_names = load_node_names(&validator_config); let bootnodes = read_bootnodes(&bootnodes_path); let validator_keys = @@ -187,7 +187,7 @@ async fn main() -> eyre::Result<()> { }) .expect("failed to build swarm"); - let p2p = P2P::spawn(built, store.clone()); + let p2p = P2P::spawn(built, store.clone(), node_names); // Wire actors together via protocol refs blockchain @@ -223,7 +223,7 @@ async fn main() -> eyre::Result<()> { Ok(()) } -fn populate_name_registry(validator_config: impl AsRef) { +fn load_node_names(validator_config: impl AsRef) -> HashMap { #[derive(Deserialize)] struct Validator { name: String, @@ -244,8 +244,9 @@ fn populate_name_registry(validator_config: impl AsRef) { .map(|v| (v.name, v.privkey)) .collect(); - // Populates a dictionary used for labeling metrics with node names - ethlambda_p2p::metrics::populate_name_registry(names_and_privkeys); + let names = ethlambda_p2p::derive_peer_ids(names_and_privkeys); + info!(count = names.len(), "Loaded node-name registry"); + names } fn read_bootnodes(bootnodes_path: impl AsRef) -> Vec { diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 58b44472..76327fcc 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -17,9 +17,9 @@ use ethrex_p2p::types::NodeRecord; use ethrex_rlp::decode::RLPDecode; use futures::StreamExt; use libp2p::{ - Multiaddr, PeerId, StreamProtocol, + Multiaddr, StreamProtocol, gossipsub::{MessageAuthenticity, ValidationMode}, - identity::{PublicKey, secp256k1}, + identity::{Keypair, PublicKey, secp256k1}, multiaddr::Protocol, request_response::{self, OutboundRequestId}, swarm::{NetworkBehaviour, SwarmEvent}, @@ -51,7 +51,7 @@ pub mod metrics; mod req_resp; pub(crate) mod swarm_adapter; -pub use metrics::populate_name_registry; +pub use libp2p::PeerId; // 5ms, 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1280ms, 2560ms const MAX_FETCH_RETRIES: u32 = 10; @@ -282,7 +282,7 @@ pub struct P2P { impl P2P { /// Build swarm, start I/O adapter, spawn actor, and wire the swarm event stream. - pub fn spawn(built: BuiltSwarm, store: Store) -> P2P { + pub fn spawn(built: BuiltSwarm, store: Store, node_names: HashMap) -> P2P { let (swarm_stream, swarm_handle) = swarm_adapter::start_swarm_adapter(built.swarm); let server = P2PServer { @@ -297,6 +297,7 @@ impl P2P { pending_requests: HashMap::new(), request_id_map: HashMap::new(), bootnode_addrs: built.bootnode_addrs, + node_names, }; let handle = server.start(); spawn_listener(handle.context(), swarm_stream.map(WrappedSwarmEvent)); @@ -332,6 +333,16 @@ pub struct P2PServer { pub(crate) pending_requests: HashMap, pub(crate) request_id_map: HashMap, bootnode_addrs: HashMap, + node_names: HashMap, +} + +impl P2PServer { + fn resolve_node_name(&self, peer_id: Option<&PeerId>) -> &str { + peer_id + .and_then(|p| self.node_names.get(p)) + .map(String::as_str) + .unwrap_or("unknown") + } } // Protocol trait for internal messages only (retry scheduling). @@ -459,7 +470,11 @@ async fn handle_swarm_event( if num_established.get() == 1 { server.connected_peers.insert(peer_id); let peer_count = server.connected_peers.len(); - metrics::notify_peer_connected(&Some(peer_id), direction, "success"); + metrics::notify_peer_connected( + server.resolve_node_name(Some(&peer_id)), + direction, + "success", + ); // Send status request on first connection to this peer let our_status = build_status(&server.store); let our_finalized_slot = our_status.finalized.slot; @@ -512,7 +527,11 @@ async fn handle_swarm_event( if num_established == 0 { server.connected_peers.remove(&peer_id); let peer_count = server.connected_peers.len(); - metrics::notify_peer_disconnected(&Some(peer_id), direction, reason); + metrics::notify_peer_disconnected( + server.resolve_node_name(Some(&peer_id)), + direction, + reason, + ); info!( %peer_id, @@ -541,7 +560,11 @@ async fn handle_swarm_event( } else { "error" }; - metrics::notify_peer_connected(&peer_id, "outbound", result); + metrics::notify_peer_connected( + server.resolve_node_name(peer_id.as_ref()), + "outbound", + result, + ); warn!(?peer_id, %error, "Outgoing connection error"); // Schedule redial if this was a bootnode @@ -558,7 +581,11 @@ async fn handle_swarm_event( } } SwarmEvent::IncomingConnectionError { peer_id, error, .. } => { - metrics::notify_peer_connected(&peer_id, "inbound", "error"); + metrics::notify_peer_connected( + server.resolve_node_name(peer_id.as_ref()), + "inbound", + "error", + ); warn!(%error, "Incoming connection error"); } _ => { @@ -567,6 +594,27 @@ async fn handle_swarm_event( } } +// --- Node identity helpers --- + +/// Drops entries whose secp256k1 private key fails to parse, with a `warn!` per drop. +pub fn derive_peer_ids(names_and_privkeys: HashMap) -> HashMap { + names_and_privkeys + .into_iter() + .filter_map(|(name, mut privkey)| { + match secp256k1::SecretKey::try_from_bytes(&mut privkey.0) { + Ok(privkey) => { + let pubkey = Keypair::from(secp256k1::Keypair::from(privkey)).public(); + Some((PeerId::from_public_key(&pubkey), name)) + } + Err(err) => { + warn!(%name, %err, "Skipping node-name registry entry: invalid secp256k1 privkey"); + None + } + } + }) + .collect() +} + // --- Bootnode parsing --- pub struct Bootnode { diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index d6d2c29b..918fd22e 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/crates/net/p2p/src/metrics.rs @@ -1,45 +1,8 @@ //! Prometheus metrics for the P2P network layer. -use std::{ - collections::HashMap, - sync::{LazyLock, RwLock}, -}; +use std::sync::LazyLock; use ethlambda_metrics::*; -use ethlambda_types::primitives::H256; -use libp2p::{ - PeerId, - identity::{Keypair, secp256k1}, -}; - -static NODE_NAME_REGISTRY: LazyLock>> = - LazyLock::new(|| RwLock::new(HashMap::new())); - -pub fn populate_name_registry(names_and_privkeys: HashMap) { - let mut registry = NODE_NAME_REGISTRY.write().unwrap(); - *registry = names_and_privkeys - .into_iter() - .filter_map(|(name, mut privkey)| { - let Ok(privkey) = secp256k1::SecretKey::try_from_bytes(&mut privkey.0) else { - return None; - }; - let pubkey = Keypair::from(secp256k1::Keypair::from(privkey)).public(); - let peer_id = PeerId::from_public_key(&pubkey); - // NOTE: we leak the name string to get a 'static lifetime. - // In reality, the name registry is not expected to be read, so it should be safe - // to turn these strings to &'static str. - Some((peer_id, &*name.leak())) - }) - .collect(); -} - -fn resolve(peer_id: &Option) -> &'static str { - let registry = NODE_NAME_REGISTRY.read().unwrap(); - peer_id - .as_ref() - .and_then(|peer_id| registry.get(peer_id)) - .unwrap_or(&"unknown") -} static LEAN_CONNECTED_PEERS: LazyLock = LazyLock::new(|| { register_int_gauge_vec!( @@ -146,25 +109,23 @@ pub fn set_attestation_committee_subnet(subnet_id: u64) { /// /// If `result` is "success", the connected peer count is incremented. /// The connection event counter is always incremented. -pub fn notify_peer_connected(peer_id: &Option, direction: &str, result: &str) { +pub fn notify_peer_connected(node_name: &str, direction: &str, result: &str) { LEAN_PEER_CONNECTION_EVENTS_TOTAL .with_label_values(&[direction, result]) .inc(); if result == "success" { - let name = resolve(peer_id); - LEAN_CONNECTED_PEERS.with_label_values(&[name]).inc(); + LEAN_CONNECTED_PEERS.with_label_values(&[node_name]).inc(); } } /// Notify that a peer disconnected. /// /// Decrements the connected peer count and increments the disconnection event counter. -pub fn notify_peer_disconnected(peer_id: &Option, direction: &str, reason: &str) { +pub fn notify_peer_disconnected(node_name: &str, direction: &str, reason: &str) { LEAN_PEER_DISCONNECTION_EVENTS_TOTAL .with_label_values(&[direction, reason]) .inc(); - let name = resolve(peer_id); - LEAN_CONNECTED_PEERS.with_label_values(&[name]).dec(); + LEAN_CONNECTED_PEERS.with_label_values(&[node_name]).dec(); } From 4cc6720b98ac6ef044aa7dbb4b628a8f4d87d214 Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 1 May 2026 22:31:27 +0300 Subject: [PATCH 2/2] Fix function doc comment --- crates/net/p2p/src/lib.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 76327fcc..a15d98f7 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -596,7 +596,9 @@ async fn handle_swarm_event( // --- Node identity helpers --- -/// Drops entries whose secp256k1 private key fails to parse, with a `warn!` per drop. +/// Derive each entry's `PeerId` from its secp256k1 private key. +/// +/// Drops entries whose key fails to parse, with a `warn!` per drop. pub fn derive_peer_ids(names_and_privkeys: HashMap) -> HashMap { names_and_privkeys .into_iter()