refactor(p2p): move node name registry into P2PServer (lambdaclass#342)

## 🗒️ Description / Motivation

The `PeerId -> node_name` lookup lived in a global
`RwLock<HashMap<PeerId, &'static str>>` populated once at startup. This
PR moves it into `P2PServer` so the actor owns its own state.

## What Changed

- **`crates/net/p2p/src/lib.rs`**: the registry is now a field on the
P2P actor's state struct, with a small helper to resolve a peer's name.
- **`crates/net/p2p/src/metrics.rs`**: no longer owns the registry — its
peer-event helpers take the resolved name directly.
- **`bin/ethlambda/src/main.rs`**: the loader now returns the map and
hands it off to `P2P::spawn` at startup; it also logs how many entries
it loaded.


## Correctness / Behavior Guarantees

- Metric label values (`lean_connected_peers{client="..."}`) are
unchanged for valid configs.
- Bad rows in `validator-config.yaml` (unparseable secp256k1 privkeys)
used to be silently dropped; they now emit a `warn!` per drop and are
still skipped. No new panics, no new failure paths.
- No consensus-layer impact (fork choice, attestations, STF, signatures
untouched).

## Tests Added / Run

  - [x] `make fmt`
  - [x] `make lint`
  - [x] `make test`

## Related Issues / PRs

  - Closes lambdaclass#106

---------

Co-authored-by: Tomás Grüner <47506558+MegaRedHand@users.noreply.github.com>

Author: conache

bin/ethlambda/src/main.rs

@@ -21,7 +21,7 @@ use tokio_util::sync::CancellationToken;
 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,
@@ -171,7 +171,7 @@ async fn main() -> eyre::Result<()> {
     );
 
     let validator_config_file = read_validator_config_file(&validator_config);
-    populate_name_registry(&validator_config_file);
+    let node_names = load_node_names(&validator_config_file);
 
     // Resolve attestation_committee_count: CLI flag > validator-config.yaml > 1.
     // The CLI path is bounded by clap's `range(1..)`; enforce the same lower
@@ -235,7 +235,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
@@ -336,7 +336,7 @@ async fn run_test_driver(rpc_config: RpcConfig) -> eyre::Result<()> {
 ///
 /// The `config` block is a network-wide settings bag shared across clients;
 /// only fields ethlambda actually reads are deserialized. The `validators`
-/// list feeds the metrics name registry.
+/// list feeds the node-name registry passed to `P2P::spawn`.
 #[derive(Debug, Deserialize)]
 struct ValidatorConfigFile {
     #[serde(default)]
@@ -361,15 +361,14 @@ fn read_validator_config_file(path: impl AsRef<Path>) -> ValidatorConfigFile {
     serde_yaml_ng::from_str(&yaml).expect("Failed to parse validator config file")
 }
 
-fn populate_name_registry(file: &ValidatorConfigFile) {
+fn load_node_names(file: &ValidatorConfigFile) -> HashMap<PeerId, String> {
     let names_and_privkeys = file
         .validators
         .iter()
         .map(|v| (v.name.clone(), v.privkey))
         .collect();
 
-    // Populates a dictionary used for labeling metrics with node names
-    ethlambda_p2p::metrics::populate_name_registry(names_and_privkeys);
+    ethlambda_p2p::derive_peer_ids(names_and_privkeys)
 }
 
 fn read_bootnodes(bootnodes_path: impl AsRef<Path>) -> Vec<Bootnode> {

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,8 +282,9 @@ 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 {
-        let (swarm_stream, swarm_handle) = swarm_adapter::start_swarm_adapter(built.swarm);
+    pub fn spawn(built: BuiltSwarm, store: Store, node_names: HashMap<PeerId, String>) -> P2P {
+        let (swarm_stream, swarm_handle) =
+            swarm_adapter::start_swarm_adapter(built.swarm, node_names.clone());
 
         let server = P2PServer {
             swarm_handle,
@@ -297,6 +298,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 +334,16 @@ pub struct P2PServer {
     pub(crate) pending_requests: HashMap<H256, PendingRequest>,
     pub(crate) request_id_map: HashMap<OutboundRequestId, H256>,
     bootnode_addrs: HashMap<PeerId, Multiaddr>,
+    node_names: HashMap<PeerId, String>,
+}
+
+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 +471,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 +528,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 +561,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 +582,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 +595,29 @@ async fn handle_swarm_event(
     }
 }
 
+// --- Node identity helpers ---
+
+/// 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<String, H256>) -> HashMap<PeerId, String> {
+    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 {

crates/net/p2p/src/metrics.rs

@@ -1,45 +1,10 @@
 //! Prometheus metrics for the P2P network layer.
 
-use std::{
-    collections::HashMap,
-    sync::{LazyLock, RwLock},
-};
+use std::collections::HashMap;
+use std::sync::LazyLock;
 
 use ethlambda_metrics::*;
-use ethlambda_types::primitives::H256;
-use libp2p::{
-    PeerId,
-    identity::{Keypair, secp256k1},
-};
-
-static NODE_NAME_REGISTRY: LazyLock<RwLock<HashMap<PeerId, &'static str>>> =
-    LazyLock::new(|| RwLock::new(HashMap::new()));
-
-pub fn populate_name_registry(names_and_privkeys: HashMap<String, H256>) {
-    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<PeerId>) -> &'static str {
-    let registry = NODE_NAME_REGISTRY.read().unwrap();
-    peer_id
-        .as_ref()
-        .and_then(|peer_id| registry.get(peer_id))
-        .unwrap_or(&"unknown")
-}
+use libp2p::PeerId;
 
 static LEAN_CONNECTED_PEERS: LazyLock<IntGaugeVec> = LazyLock::new(|| {
     register_int_gauge_vec!(
@@ -237,38 +202,39 @@ 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<PeerId>, 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<PeerId>, 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();
 }
 
 /// Refresh the gossipsub mesh peers gauge from the current mesh peer set.
-pub fn update_gossip_mesh_peers<'a>(peers: impl Iterator<Item = &'a PeerId>) {
+pub fn update_gossip_mesh_peers<'a>(
+    peers: impl Iterator<Item = &'a PeerId>,
+    node_names: &HashMap<PeerId, String>,
+) {
     let mut counts: HashMap<String, i64> = HashMap::new();
-    {
-        let registry = NODE_NAME_REGISTRY.read().unwrap();
-        for peer_id in peers {
-            let name = registry.get(peer_id).copied().unwrap_or("unknown");
-            *counts.entry(name.to_string()).or_default() += 1;
-        }
+    for peer_id in peers {
+        let name = node_names
+            .get(peer_id)
+            .map(String::as_str)
+            .unwrap_or("unknown");
+        *counts.entry(name.to_string()).or_default() += 1;
     }
     // Seed previously-published labels with 0 so departed clients fall to
     // zero in the single set() pass below.

crates/net/p2p/src/swarm_adapter.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::time::Duration;
 
 use libp2p::{
@@ -92,14 +93,15 @@ impl SwarmHandle {
 
 pub fn start_swarm_adapter(
     swarm: libp2p::Swarm<Behaviour>,
+    node_names: HashMap<PeerId, String>,
 ) -> (
     impl futures::Stream<Item = SwarmEvent<BehaviourEvent>>,
     SwarmHandle,
 ) {
     let (event_tx, event_rx) = mpsc::unbounded_channel();
     let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
 
-    tokio::spawn(swarm_loop(swarm, event_tx, cmd_rx));
+    tokio::spawn(swarm_loop(swarm, event_tx, cmd_rx, node_names));
 
     let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(event_rx);
     let handle = SwarmHandle { cmd_tx };
@@ -110,6 +112,7 @@ async fn swarm_loop(
     mut swarm: libp2p::Swarm<Behaviour>,
     event_tx: mpsc::UnboundedSender<SwarmEvent<BehaviourEvent>>,
     mut cmd_rx: mpsc::UnboundedReceiver<SwarmCommand>,
+    node_names: HashMap<PeerId, String>,
 ) {
     let mut mesh_metric_tick = tokio::time::interval(MESH_METRIC_REFRESH_INTERVAL);
     mesh_metric_tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
@@ -126,6 +129,7 @@ async fn swarm_loop(
             _ = mesh_metric_tick.tick() => {
                 metrics::update_gossip_mesh_peers(
                     swarm.behaviour().gossipsub.all_mesh_peers(),
+                    &node_names,
                 );
             }
         }