use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use async_trait::async_trait;
use parking_lot::Mutex;
use codec::Encode;
use sc_network::{
config::parse_addr, multiaddr::Multiaddr, service::traits::NetworkService, types::ProtocolName,
IfDisconnected, MessageSink, OutboundFailure, ReputationChange, RequestFailure,
};
use polkadot_node_network_protocol::{
peer_set::{CollationVersion, PeerSet, ProtocolVersion, ValidationVersion},
request_response::{OutgoingRequest, Recipient, ReqProtocolNames, Requests},
v1 as protocol_v1, v2 as protocol_v2, v3 as protocol_v3, PeerId,
};
use polkadot_primitives::AuthorityDiscoveryId;
use crate::{metrics::Metrics, validator_discovery::AuthorityDiscovery, WireMessage};
const LOG_TARGET: &'static str = "parachain::network-bridge-net";
pub(crate) fn send_validation_message_v1(
peers: Vec<PeerId>,
message: WireMessage<protocol_v1::ValidationProtocol>,
metrics: &Metrics,
notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) {
gum::trace!(target: LOG_TARGET, ?peers, ?message, "Sending validation v1 message to peers",);
send_message(
peers,
PeerSet::Validation,
ValidationVersion::V1.into(),
message,
metrics,
notification_sinks,
);
}
pub(crate) fn send_validation_message_v3(
peers: Vec<PeerId>,
message: WireMessage<protocol_v3::ValidationProtocol>,
metrics: &Metrics,
notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) {
gum::trace!(target: LOG_TARGET, ?peers, ?message, "Sending validation v3 message to peers",);
send_message(
peers,
PeerSet::Validation,
ValidationVersion::V3.into(),
message,
metrics,
notification_sinks,
);
}
pub(crate) fn send_validation_message_v2(
peers: Vec<PeerId>,
message: WireMessage<protocol_v2::ValidationProtocol>,
metrics: &Metrics,
notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) {
send_message(
peers,
PeerSet::Validation,
ValidationVersion::V2.into(),
message,
metrics,
notification_sinks,
);
}
pub(crate) fn send_collation_message_v1(
peers: Vec<PeerId>,
message: WireMessage<protocol_v1::CollationProtocol>,
metrics: &Metrics,
notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) {
send_message(
peers,
PeerSet::Collation,
CollationVersion::V1.into(),
message,
metrics,
notification_sinks,
);
}
pub(crate) fn send_collation_message_v2(
peers: Vec<PeerId>,
message: WireMessage<protocol_v2::CollationProtocol>,
metrics: &Metrics,
notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) {
send_message(
peers,
PeerSet::Collation,
CollationVersion::V2.into(),
message,
metrics,
notification_sinks,
);
}
fn send_message<M>(
mut peers: Vec<PeerId>,
peer_set: PeerSet,
version: ProtocolVersion,
message: M,
metrics: &super::Metrics,
network_notification_sinks: &Arc<Mutex<HashMap<(PeerSet, PeerId), Box<dyn MessageSink>>>>,
) where
M: Encode + Clone,
{
if peers.is_empty() {
return
}
let message = {
let encoded = message.encode();
metrics.on_notification_sent(peer_set, version, encoded.len(), peers.len());
metrics.on_message(std::any::type_name::<M>());
encoded
};
let notification_sinks = network_notification_sinks.lock();
gum::trace!(
target: LOG_TARGET,
?peers,
?peer_set,
?version,
?message,
"Sending message to peers",
);
let last_peer = peers.pop();
peers.into_iter().for_each(|peer| {
if let Some(sink) = notification_sinks.get(&(peer_set, peer)) {
sink.send_sync_notification(message.clone());
}
});
if let Some(peer) = last_peer {
if let Some(sink) = notification_sinks.get(&(peer_set, peer)) {
sink.send_sync_notification(message.clone());
}
}
}
#[async_trait]
pub trait Network: Clone + Send + 'static {
async fn set_reserved_peers(
&mut self,
protocol: ProtocolName,
multiaddresses: HashSet<Multiaddr>,
) -> Result<(), String>;
async fn add_peers_to_reserved_set(
&mut self,
protocol: ProtocolName,
multiaddresses: HashSet<Multiaddr>,
) -> Result<(), String>;
async fn remove_from_peers_set(
&mut self,
protocol: ProtocolName,
peers: Vec<PeerId>,
) -> Result<(), String>;
async fn start_request<AD: AuthorityDiscovery>(
&self,
authority_discovery: &mut AD,
req: Requests,
req_protocol_names: &ReqProtocolNames,
if_disconnected: IfDisconnected,
);
fn report_peer(&self, who: PeerId, rep: ReputationChange);
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName);
fn peer_role(&self, who: PeerId, handshake: Vec<u8>) -> Option<sc_network::ObservedRole>;
}
#[async_trait]
impl Network for Arc<dyn NetworkService> {
async fn set_reserved_peers(
&mut self,
protocol: ProtocolName,
multiaddresses: HashSet<Multiaddr>,
) -> Result<(), String> {
<dyn NetworkService>::set_reserved_peers(&**self, protocol, multiaddresses)
}
async fn add_peers_to_reserved_set(
&mut self,
protocol: ProtocolName,
multiaddresses: HashSet<Multiaddr>,
) -> Result<(), String> {
<dyn NetworkService>::add_peers_to_reserved_set(&**self, protocol, multiaddresses)
}
async fn remove_from_peers_set(
&mut self,
protocol: ProtocolName,
peers: Vec<PeerId>,
) -> Result<(), String> {
<dyn NetworkService>::remove_peers_from_reserved_set(&**self, protocol, peers)
}
fn report_peer(&self, who: PeerId, rep: ReputationChange) {
<dyn NetworkService>::report_peer(&**self, who, rep);
}
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
<dyn NetworkService>::disconnect_peer(&**self, who, protocol);
}
async fn start_request<AD: AuthorityDiscovery>(
&self,
authority_discovery: &mut AD,
req: Requests,
req_protocol_names: &ReqProtocolNames,
if_disconnected: IfDisconnected,
) {
let (protocol, OutgoingRequest { peer, payload, pending_response, fallback_request }) =
req.encode_request();
let peer_id = match peer {
Recipient::Peer(peer_id) => Some(peer_id),
Recipient::Authority(authority) => {
gum::trace!(
target: LOG_TARGET,
?authority,
"Searching for peer id to connect to authority",
);
let mut found_peer_id = None;
for addr in authority_discovery
.get_addresses_by_authority_id(authority)
.await
.into_iter()
.flat_map(|list| list.into_iter())
{
let (peer_id, addr) = match parse_addr(addr) {
Ok(v) => v,
Err(_) => continue,
};
<dyn NetworkService>::add_known_address(&**self, peer_id, addr);
found_peer_id = Some(peer_id);
}
found_peer_id
},
};
let peer_id = match peer_id {
None => {
gum::debug!(target: LOG_TARGET, "Discovering authority failed");
match pending_response
.send(Err(RequestFailure::Network(OutboundFailure::DialFailure)))
{
Err(_) => {
gum::debug!(target: LOG_TARGET, "Sending failed request response failed.")
},
Ok(_) => {},
}
return
},
Some(peer_id) => peer_id,
};
gum::trace!(
target: LOG_TARGET,
%peer_id,
protocol = %req_protocol_names.get_name(protocol),
fallback_protocol = ?fallback_request.as_ref().map(|(_, p)| req_protocol_names.get_name(*p)),
?if_disconnected,
"Starting request",
);
<dyn NetworkService>::start_request(
&**self,
peer_id,
req_protocol_names.get_name(protocol),
payload,
fallback_request.map(|(r, p)| (r, req_protocol_names.get_name(p))),
pending_response,
if_disconnected,
);
}
fn peer_role(&self, who: PeerId, handshake: Vec<u8>) -> Option<sc_network::ObservedRole> {
<dyn NetworkService>::peer_role(&**self, who, handshake)
}
}
pub async fn get_peer_id_by_authority_id<AD: AuthorityDiscovery>(
authority_discovery: &mut AD,
authority: AuthorityDiscoveryId,
) -> Option<PeerId> {
authority_discovery
.get_addresses_by_authority_id(authority)
.await
.into_iter()
.flat_map(|list| list.into_iter())
.find_map(|addr| parse_addr(addr).ok().map(|(p, _)| p))
}