statement-store: add event-fed v2 peer topology

substrate/client/network/src/event.rs

@@ -22,6 +22,7 @@
 use crate::types::ProtocolName;
 
 use bytes::Bytes;
+use std::collections::HashSet;
 
 use sc_network_common::role::ObservedRole;
 use sc_network_types::{
@@ -81,6 +82,17 @@ pub enum Event {
 	/// Event generated by a DHT.
 	Dht(DhtEvent),
 
+	/// The network discovered a peer through routing-table/discovery mechanisms.
+	PeerDiscovered(PeerId),
+
+	/// The network identified a peer and learned the protocols it supports.
+	PeerIdentified {
+		/// Identified peer.
+		peer: PeerId,
+		/// Protocols reported by the peer.
+		supported_protocols: HashSet<ProtocolName>,
+	},
+
 	/// Opened a substream with the given node with the given notifications protocol.
 	///
 	/// The protocol is always one of the notification protocols that have been registered.

substrate/client/network/src/litep2p/mod.rs

@@ -816,7 +816,9 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkBackend<B, H> for Litep2pNetworkBac
 					}
 					Some(DiscoveryEvent::RoutingTableUpdate { peers }) => {
 						for peer in peers {
-							self.peerstore_handle.add_known_peer(peer.into());
+							let peer = peer.into();
+							self.peerstore_handle.add_known_peer(peer);
+							self.event_streams.send(Event::PeerDiscovered(peer));
 						}
 					}
 					Some(DiscoveryEvent::FindNodeSuccess { query_id, target, peers }) => {
@@ -1119,6 +1121,10 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkBackend<B, H> for Litep2pNetworkBac
 						}
 					}
 					Some(DiscoveryEvent::Identified { peer, listen_addresses, supported_protocols, .. }) => {
+						self.event_streams.send(Event::PeerIdentified {
+							peer: peer.into(),
+							supported_protocols: supported_protocols.iter().cloned().map(Into::into).collect(),
+						});
 						self.discovery.add_self_reported_address(peer, supported_protocols, listen_addresses).await;
 					}
 					Some(DiscoveryEvent::ExternalAddressDiscovered { address }) => {

substrate/client/network/src/service.rs

@@ -1617,10 +1617,20 @@ where
 						addr.clone(),
 					);
 				}
-				self.peer_store_handle.add_known_peer(peer_id.into());
+				let peer = peer_id.into();
+				self.peer_store_handle.add_known_peer(peer);
+				self.event_streams.send(Event::PeerIdentified {
+					peer,
+					supported_protocols: protocols
+						.into_iter()
+						.map(|protocol| ProtocolName::from(protocol.to_string()))
+						.collect(),
+				});
 			},
 			SwarmEvent::Behaviour(BehaviourOut::Discovered(peer_id)) => {
-				self.peer_store_handle.add_known_peer(peer_id.into());
+				let peer = peer_id.into();
+				self.peer_store_handle.add_known_peer(peer);
+				self.event_streams.send(Event::PeerDiscovered(peer));
 			},
 			SwarmEvent::Behaviour(BehaviourOut::RandomKademliaStarted) => {
 				if let Some(metrics) = self.metrics.as_ref() {

substrate/client/network/src/service/out_events.rs

@@ -294,6 +294,12 @@ impl Metrics {
 			Event::Dht(_) => {
 				self.events_total.with_label_values(&["dht", "sent", name]).inc();
 			},
+			Event::PeerDiscovered(_) => {
+				self.events_total.with_label_values(&["peer-discovered", "sent", name]).inc();
+			},
+			Event::PeerIdentified { .. } => {
+				self.events_total.with_label_values(&["peer-identified", "sent", name]).inc();
+			},
 			Event::NotificationStreamOpened { protocol, .. } => {
 				format_label("notif-open-", protocol, |protocol_label| {
 					self.events_total.with_label_values(&[protocol_label, "sent", name]).inc();
@@ -322,6 +328,16 @@ impl Metrics {
 			Event::Dht(_) => {
 				self.events_total.with_label_values(&["dht", "received", name]).inc();
 			},
+			Event::PeerDiscovered(_) => {
+				self.events_total
+					.with_label_values(&["peer-discovered", "received", name])
+					.inc();
+			},
+			Event::PeerIdentified { .. } => {
+				self.events_total
+					.with_label_values(&["peer-identified", "received", name])
+					.inc();
+			},
 			Event::NotificationStreamOpened { protocol, .. } => {
 				format_label("notif-open-", protocol, |protocol_label| {
 					self.events_total.with_label_values(&[protocol_label, "received", name]).inc();

substrate/client/network/statement/Cargo.toml

@@ -33,6 +33,7 @@ sc-network-common = { workspace = true, default-features = true }
 sc-network-sync = { workspace = true, default-features = true }
 sc-network-types = { workspace = true, default-features = true }
 sp-consensus = { workspace = true, default-features = true }
+sp-crypto-hashing = { workspace = true, default-features = true }
 sp-runtime = { workspace = true, default-features = true }
 sp-statement-store = { workspace = true, default-features = true }
 tokio = { workspace = true }
@@ -45,7 +46,6 @@ sc-statement-store = { workspace = true, default-features = true }
 sp-api = { workspace = true, default-features = true }
 sp-blockchain = { workspace = true, default-features = true }
 sp-core = { workspace = true, default-features = true }
-sp-crypto-hashing = { workspace = true, default-features = true }
 sp-keyring = { workspace = true, default-features = true }
 substrate-test-runtime-client = { workspace = true }
 tempfile = { workspace = true }