feat(rpc): serve /eth/v1/node/peers, /peers/{peer_id} and /peer_count

bin/ethlambda/src/main.rs

@@ -152,6 +152,7 @@ fn beacon_identity(
         ],
         discovery_addresses: vec![format!("/{family}/{ip}/udp/{discovery_port}/p2p/{peer_id}")],
         metadata: None,
+        peers: None,
     }
 }
 
@@ -810,6 +811,7 @@ async fn run_node(options: Options) -> eyre::Result<()> {
     };
     let rpc_identity = ethlambda_rpc::BeaconIdentity {
         metadata: p2p.local_metadata(),
+        peers: Some(p2p.peer_table()),
         ..beacon_identity(
             p2p.local_enr().unwrap_or_default(),
             common.discovery.advertise_ip,

crates/net/api/src/lib.rs

@@ -515,6 +515,138 @@ impl MetaDataController {
     }
 }
 
+/// Which side opened a peer's first connection.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PeerDirection {
+    /// The remote dialed us.
+    Inbound,
+    /// We dialed the remote.
+    Outbound,
+}
+
+impl PeerDirection {
+    /// The value the Beacon API's `Peer.direction` carries.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Inbound => "inbound",
+            Self::Outbound => "outbound",
+        }
+    }
+}
+
+/// One connected peer, as the Beacon API's `/eth/v1/node/peers` reports it.
+///
+/// Strings rather than libp2p types, so this crate stays free of libp2p.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ConnectedPeer {
+    /// The peer's libp2p id, base58.
+    pub peer_id: String,
+    pub direction: PeerDirection,
+    /// The remote address of the connection that made the peer connected, ending
+    /// in `/p2p/<peer id>`.
+    pub last_seen_p2p_address: String,
+    /// The peer's `enr:`-prefixed record, when discovery supplied one before we
+    /// dialed it. `None` for an inbound peer, whose record we never fetch.
+    pub enr: Option<String>,
+}
+
+/// Shared, runtime-readable table of the peers this node is connected to.
+///
+/// The p2p layer is the sole writer: it inserts a peer when its first
+/// connection is established and removes it when its last one closes, so only
+/// connected peers are ever in the table. The Beacon API's
+/// `/eth/v1/node/peers` clones the handle and reads it at request time.
+#[derive(Clone, Debug, Default)]
+pub struct PeerTableController {
+    peers: Arc<RwLock<std::collections::HashMap<String, ConnectedPeer>>>,
+}
+
+impl PeerTableController {
+    /// Record a peer as connected, replacing any earlier entry for it.
+    pub fn insert(&self, peer: ConnectedPeer) {
+        self.peers
+            .write()
+            .expect("peer table lock poisoned")
+            .insert(peer.peer_id.clone(), peer);
+    }
+
+    /// Drop a peer that no longer holds a connection.
+    pub fn remove(&self, peer_id: &str) {
+        self.peers
+            .write()
+            .expect("peer table lock poisoned")
+            .remove(peer_id);
+    }
+
+    /// One connected peer, if it is.
+    pub fn get(&self, peer_id: &str) -> Option<ConnectedPeer> {
+        self.peers
+            .read()
+            .expect("peer table lock poisoned")
+            .get(peer_id)
+            .cloned()
+    }
+
+    /// Every connected peer, ordered by peer id so a response is stable
+    /// between polls.
+    pub fn snapshot(&self) -> Vec<ConnectedPeer> {
+        let mut peers: Vec<_> = self
+            .peers
+            .read()
+            .expect("peer table lock poisoned")
+            .values()
+            .cloned()
+            .collect();
+        peers.sort_by(|a, b| a.peer_id.cmp(&b.peer_id));
+        peers
+    }
+}
+
+#[cfg(test)]
+mod peer_table_tests {
+    use super::*;
+
+    fn peer(id: &str, direction: PeerDirection) -> ConnectedPeer {
+        ConnectedPeer {
+            peer_id: id.to_string(),
+            direction,
+            last_seen_p2p_address: format!("/ip4/10.0.0.1/tcp/9000/p2p/{id}"),
+            enr: None,
+        }
+    }
+
+    #[test]
+    fn a_peer_is_present_from_insert_until_remove() {
+        let table = PeerTableController::default();
+        table.insert(peer("b", PeerDirection::Inbound));
+        assert_eq!(table.get("b").unwrap().direction, PeerDirection::Inbound);
+
+        table.remove("b");
+        assert!(table.get("b").is_none());
+        assert!(table.snapshot().is_empty());
+    }
+
+    #[test]
+    fn inserting_twice_keeps_one_entry_with_the_latest_value() {
+        let table = PeerTableController::default();
+        table.insert(peer("a", PeerDirection::Inbound));
+        table.insert(peer("a", PeerDirection::Outbound));
+        let peers = table.snapshot();
+        assert_eq!(peers.len(), 1);
+        assert_eq!(peers[0].direction, PeerDirection::Outbound);
+    }
+
+    #[test]
+    fn the_snapshot_is_ordered_by_peer_id_and_shared_between_clones() {
+        let table = PeerTableController::default();
+        let reader = table.clone();
+        table.insert(peer("c", PeerDirection::Outbound));
+        table.insert(peer("a", PeerDirection::Inbound));
+        let ids: Vec<_> = reader.snapshot().into_iter().map(|p| p.peer_id).collect();
+        assert_eq!(ids, ["a", "c"]);
+    }
+}
+
 // --- Init messages ---
 // Used to wire actors together after spawn.
 

crates/net/p2p/src/discovery/admission.rs

@@ -54,6 +54,9 @@ pub(crate) struct DiscoveredPeer {
     /// `P2PServer::record_peer_custody`), and this only fills the gap until
     /// that answer arrives. Lighthouse splits the two the same way.
     pub(crate) custody_group_count: Option<u64>,
+    /// The record as an `enr:` string, kept for the Beacon API's peer list.
+    /// `None` only if the record cannot be re-encoded.
+    pub(crate) enr: Option<String>,
 }
 
 /// Why a discovered peer was turned away.
@@ -235,6 +238,7 @@ fn admit(
         addrs: dial_addrs(ip, quic_port, tcp_port, peer_id),
         subnets,
         custody_group_count,
+        enr: record.enr_url().ok(),
     })
 }
 
@@ -396,6 +400,7 @@ mod tests {
                 addrs: vec![Multiaddr::empty()],
                 subnets,
                 custody_group_count: None,
+                enr: None,
             }
         }
     }
@@ -730,6 +735,7 @@ mod tests {
             addrs: vec![Multiaddr::empty()],
             subnets: vec![],
             custody_group_count: Some(custody_group_count),
+            enr: None,
         }
     }
 

crates/net/p2p/src/discovery/dial.rs

@@ -36,6 +36,10 @@ pub(crate) struct DiscoveryState {
     candidates: VecDeque<DiscoveredPeer>,
     /// Subnets advertised by peers we dialed from discovery.
     peer_attnets: HashMap<PeerId, Vec<u64>>,
+    /// The ENR of each peer we dialed from discovery, until its connection
+    /// ends or its dial fails. Read once, when the connection is established,
+    /// for the Beacon API's peer list.
+    peer_enrs: HashMap<PeerId, String>,
     /// Our own peer ID, so the loop never dials itself.
     local_peer_id: PeerId,
     /// Connected-peer count above which the loop stops dialing.
@@ -53,6 +57,7 @@ impl DiscoveryState {
             contacts: spawn_contact_poll(handle.peer_table, handle.filter.clone()),
             candidates: VecDeque::new(),
             peer_attnets: HashMap::new(),
+            peer_enrs: HashMap::new(),
             local_peer_id,
             target_peers: handle.target_peers,
             filter: handle.filter,
@@ -68,6 +73,11 @@ impl DiscoveryState {
     pub(crate) fn target_peers(&self) -> usize {
         self.target_peers
     }
+
+    /// The ENR discovery supplied for a peer we dialed, if any.
+    pub(crate) fn enr_of(&self, peer_id: &PeerId) -> Option<String> {
+        self.peer_enrs.get(peer_id).cloned()
+    }
 }
 
 /// How long the poll task waits after the peer table offers nothing.
@@ -93,6 +103,7 @@ const CONTACT_POLL_IDLE: Duration = Duration::from_millis(250);
 pub(crate) fn forget_discovered_peer(server: &mut P2PServer, peer_id: &PeerId) {
     if let Some(discovery) = server.discovery.as_mut() {
         discovery.peer_attnets.remove(peer_id);
+        discovery.peer_enrs.remove(peer_id);
     }
     // Custody is keyed by peer id and a peer id is a public key, so a returning
     // peer recomputes to the same set. Dropped anyway: the map is only ever
@@ -328,6 +339,9 @@ pub(crate) async fn dial_tick(server: &mut P2PServer, target_peers: usize) -> bo
             discovery
                 .peer_attnets
                 .insert(candidate.peer_id, candidate.subnets);
+            if let Some(enr) = candidate.enr {
+                discovery.peer_enrs.insert(candidate.peer_id, enr);
+            }
         }
     }
     dialed

crates/net/p2p/src/lib.rs

@@ -38,7 +38,8 @@ use std::{
 
 use either::Either;
 use ethlambda_network_api::{
-    FetchRequest, InitBlockChain, MetaDataController, P2PToBlockChainRef,
+    ConnectedPeer, FetchRequest, InitBlockChain, MetaDataController, P2PToBlockChainRef,
+    PeerDirection, PeerTableController,
     block_chain_to_p2p::{
         CheckDataColumnSidecars, FetchBlock, PublishAggregatedAttestation, PublishAttestation,
         PublishBlock,
@@ -1154,6 +1155,9 @@ pub struct P2P {
     /// The beacon `MetaData` this node answers peers with, for the same
     /// endpoint. `None` on a lean wire, which has no `MetaData`.
     local_metadata: Option<MetaDataController>,
+    /// The peers this node is connected to, for the Beacon API's
+    /// `/eth/v1/node/peers`.
+    peer_table: PeerTableController,
 }
 
 impl P2P {
@@ -1212,13 +1216,15 @@ impl P2P {
             .beacon()
             .map(|beacon| beacon::scoring::PeerScoring::new(&beacon.config, beacon.fork_digest));
         let local_metadata = built.wire.beacon().map(|beacon| beacon.metadata.clone());
+        let peer_table = PeerTableController::default();
 
         let server = P2PServer {
             swarm_handle,
             store,
             blockchain: None,
             wire: built.wire,
             connected_peers: HashMap::new(),
+            peer_table: peer_table.clone(),
             peer_custody: HashMap::new(),
             pending_root_requests: HashMap::new(),
             pending_envelope_requests: HashMap::new(),
@@ -1303,6 +1309,7 @@ impl P2P {
             handle,
             local_enr,
             local_metadata,
+            peer_table,
         })
     }
 
@@ -1323,6 +1330,12 @@ impl P2P {
     pub fn local_metadata(&self) -> Option<MetaDataController> {
         self.local_metadata.clone()
     }
+
+    /// A live view of the peers this node is connected to. Written only by
+    /// the P2P actor.
+    pub fn peer_table(&self) -> PeerTableController {
+        self.peer_table.clone()
+    }
 }
 
 /// Message wrapper for swarm events. Not part of the protocol because
@@ -1351,6 +1364,9 @@ pub struct P2PServer {
     /// direction is the first connection's, which is the one that decides
     /// whether this peer was our choice or the network's.
     pub(crate) connected_peers: HashMap<PeerId, ConnectionDirection>,
+    /// [`Self::connected_peers`] plus what the Beacon API reports about each
+    /// peer, shared with the RPC layer. This actor is its only writer.
+    pub(crate) peer_table: PeerTableController,
     /// The columns each peer custodies, as computed from its own node id and
     /// its advertised custody group count.
     ///
@@ -1503,6 +1519,40 @@ impl P2PServer {
         self.refresh_custody_column_metrics();
     }
 
+    /// Publish a newly connected peer to the table the Beacon API reads.
+    ///
+    /// `remote` is the address of the connection that made the peer connected;
+    /// `/p2p/<peer id>` is appended unless the transport already carries it.
+    /// The ENR is whatever the dial loop kept from discovery, so it is absent
+    /// for an inbound peer and for a bootnode dialed from `--bootnodes`.
+    fn publish_connected_peer(
+        &self,
+        peer_id: PeerId,
+        direction: ConnectionDirection,
+        remote: &Multiaddr,
+    ) {
+        let mut address = remote.clone();
+        if !address
+            .iter()
+            .any(|protocol| matches!(protocol, Protocol::P2p(_)))
+        {
+            address.push(Protocol::P2p(peer_id));
+        }
+        let direction = match direction {
+            ConnectionDirection::Inbound => PeerDirection::Inbound,
+            ConnectionDirection::Outbound => PeerDirection::Outbound,
+        };
+        self.peer_table.insert(ConnectedPeer {
+            peer_id: peer_id.to_string(),
+            direction,
+            last_seen_p2p_address: address.to_string(),
+            enr: self
+                .discovery
+                .as_ref()
+                .and_then(|discovery| discovery.enr_of(&peer_id)),
+        });
+    }
+
     /// Publish, per column this node samples, how many connected peers are
     /// known to custody it.
     ///
@@ -1971,6 +2021,7 @@ async fn handle_swarm_event(
             metrics::inc_peer_connection_transport(direction.as_str(), transport);
             if num_established.get() == 1 {
                 server.connected_peers.insert(peer_id, direction);
+                server.publish_connected_peer(peer_id, direction, endpoint.get_remote_address());
                 server.refresh_peer_metrics();
                 let peer_count = server.connected_peers.len();
                 metrics::notify_peer_connected(
@@ -2071,6 +2122,7 @@ async fn handle_swarm_event(
                     .connected_peers
                     .remove(&peer_id)
                     .unwrap_or(closed_direction);
+                server.peer_table.remove(&peer_id.to_string());
                 forget_discovered_peer(server, &peer_id);
                 server.refresh_peer_metrics();
                 let peer_count = server.connected_peers.len();
@@ -3035,6 +3087,7 @@ pub(crate) mod test_support {
             blockchain: None,
             wire: built.wire,
             connected_peers: HashMap::new(),
+            peer_table: ethlambda_network_api::PeerTableController::default(),
             peer_custody: HashMap::new(),
             pending_root_requests: HashMap::new(),
             pending_envelope_requests: HashMap::new(),

crates/net/p2p/src/req_resp/handlers.rs

@@ -2773,6 +2773,7 @@ pub(crate) mod tests {
             blockchain: None,
             wire: built.wire,
             connected_peers: HashMap::new(),
+            peer_table: ethlambda_network_api::PeerTableController::default(),
             peer_custody: HashMap::new(),
             pending_root_requests: HashMap::new(),
             pending_envelope_requests: HashMap::new(),