use crate::{
ExHashT,
chain::Client,
config::{self, ProtocolId, TransactionPool, TransactionImportFuture, TransactionImport},
error,
request_responses::RequestFailure,
utils::{interval, LruHashSet},
};
use bytes::{Bytes, BytesMut};
use codec::{Decode, DecodeAll, Encode};
use futures::{channel::oneshot, prelude::*, stream::FuturesUnordered};
use generic_proto::{GenericProto, GenericProtoOut};
use libp2p::core::{ConnectedPoint, connection::{ConnectionId, ListenerId}};
use libp2p::request_response::OutboundFailure;
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::swarm::{ProtocolsHandler, IntoProtocolsHandler};
use libp2p::{Multiaddr, PeerId};
use log::{log, Level, trace, debug, warn, error};
use message::{BlockAnnounce, Message};
use message::generic::{Message as GenericMessage, Roles};
use prometheus_endpoint::{
Registry, Gauge, Counter, GaugeVec,
PrometheusError, Opts, register, U64
};
use prost::Message as _;
use sp_consensus::{
BlockOrigin,
block_validation::BlockAnnounceValidator,
import_queue::{BlockImportResult, BlockImportError, IncomingBlock, Origin}
};
use sp_runtime::{generic::BlockId, Justification};
use sp_runtime::traits::{
Block as BlockT, Header as HeaderT, NumberFor, Zero, CheckedSub
};
use sp_arithmetic::traits::SaturatedConversion;
use sync::{ChainSync, SyncState};
use std::borrow::Cow;
use std::convert::TryFrom as _;
use std::collections::{HashMap, HashSet, VecDeque, hash_map::Entry};
use std::sync::Arc;
use std::{io, iter, num::NonZeroUsize, pin::Pin, task::Poll, time};
mod generic_proto;
pub mod message;
pub mod event;
pub mod sync;
pub use generic_proto::{NotificationsSink, Ready, NotifsHandlerError};
const TICK_TIMEOUT: time::Duration = time::Duration::from_millis(1100);
const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis(2900);
const MAX_KNOWN_BLOCKS: usize = 1024;
const MAX_KNOWN_TRANSACTIONS: usize = 10240;
const MAX_BLOCK_ANNOUNCE_SIZE: u64 = 1024 * 1024;
const MAX_TRANSACTIONS_SIZE: u64 = 16 * 1024 * 1024;
pub(crate) const BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE: u64 = 16 * 1024 * 1024;
const MAX_PENDING_TRANSACTIONS: usize = 8192;
pub(crate) const CURRENT_VERSION: u32 = 6;
pub(crate) const MIN_VERSION: u32 = 3;
const HARDCODED_PEERSETS_SYNC: sc_peerset::SetId = sc_peerset::SetId::from(0);
const HARDCODED_PEERSETS_TX: sc_peerset::SetId = sc_peerset::SetId::from(1);
const NUM_HARDCODED_PEERSETS: usize = 2;
const LIGHT_MAXIMAL_BLOCKS_DIFFERENCE: u64 = 8192;
mod rep {
use sc_peerset::ReputationChange as Rep;
pub const TIMEOUT: Rep = Rep::new(-(1 << 10), "Request timeout");
pub const REFUSED: Rep = Rep::new(-(1 << 10), "Request refused");
pub const PEER_BEHIND_US_LIGHT: Rep = Rep::new(-(1 << 8), "Useless for a light peer");
pub const ANY_TRANSACTION: Rep = Rep::new(-(1 << 4), "Any transaction");
pub const ANY_TRANSACTION_REFUND: Rep = Rep::new(1 << 4, "Any transaction (refund)");
pub const GOOD_TRANSACTION: Rep = Rep::new(1 << 7, "Good transaction");
pub const BAD_TRANSACTION: Rep = Rep::new(-(1 << 12), "Bad transaction");
pub const BAD_MESSAGE: Rep = Rep::new(-(1 << 12), "Bad message");
pub const UNEXPECTED_TRANSACTIONS: Rep = Rep::new_fatal("Unexpected transactions packet");
pub const GENESIS_MISMATCH: Rep = Rep::new_fatal("Genesis mismatch");
pub const BAD_PROTOCOL: Rep = Rep::new_fatal("Unsupported protocol");
pub const BAD_ROLE: Rep = Rep::new_fatal("Unsupported role");
pub const BAD_BLOCK_ANNOUNCEMENT: Rep = Rep::new(-(1 << 12), "Bad block announcement");
}
struct Metrics {
peers: Gauge<U64>,
queued_blocks: Gauge<U64>,
fork_targets: Gauge<U64>,
justifications: GaugeVec<U64>,
propagated_transactions: Counter<U64>,
}
impl Metrics {
fn register(r: &Registry) -> Result<Self, PrometheusError> {
Ok(Metrics {
peers: {
let g = Gauge::new("sync_peers", "Number of peers we sync with")?;
register(g, r)?
},
queued_blocks: {
let g = Gauge::new("sync_queued_blocks", "Number of blocks in import queue")?;
register(g, r)?
},
fork_targets: {
let g = Gauge::new("sync_fork_targets", "Number of fork sync targets")?;
register(g, r)?
},
justifications: {
let g = GaugeVec::new(
Opts::new(
"sync_extra_justifications",
"Number of extra justifications requests"
),
&["status"],
)?;
register(g, r)?
},
propagated_transactions: register(Counter::new(
"sync_propagated_transactions",
"Number of transactions propagated to at least one peer",
)?, r)?,
})
}
}
#[pin_project::pin_project]
struct PendingTransaction<H> {
#[pin]
validation: TransactionImportFuture,
tx_hash: H,
}
impl<H: ExHashT> Future for PendingTransaction<H> {
type Output = (H, TransactionImport);
fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
if let Poll::Ready(import_result) = Pin::new(&mut this.validation).poll_unpin(cx) {
return Poll::Ready((this.tx_hash.clone(), import_result));
}
Poll::Pending
}
}
pub struct Protocol<B: BlockT, H: ExHashT> {
tick_timeout: Pin<Box<dyn Stream<Item = ()> + Send>>,
propagate_timeout: Pin<Box<dyn Stream<Item = ()> + Send>>,
pending_messages: VecDeque<CustomMessageOutcome<B>>,
pending_transactions: FuturesUnordered<PendingTransaction<H>>,
pending_transactions_peers: HashMap<H, Vec<PeerId>>,
config: ProtocolConfig,
genesis_hash: B::Hash,
sync: ChainSync<B>,
peers: HashMap<PeerId, Peer<B, H>>,
chain: Arc<dyn Client<B>>,
important_peers: HashSet<PeerId>,
peerset_handle: sc_peerset::PeersetHandle,
transaction_pool: Arc<dyn TransactionPool<H, B>>,
behaviour: GenericProto,
notification_protocols: Vec<Cow<'static, str>>,
metrics: Option<Metrics>,
boot_node_ids: HashSet<PeerId>,
block_announce_data_cache: lru::LruCache<B::Hash, Vec<u8>>,
}
#[derive(Debug)]
struct Peer<B: BlockT, H: ExHashT> {
info: PeerInfo<B>,
block_request: Option<(
message::BlockRequest<B>,
oneshot::Receiver<Result<Vec<u8>, RequestFailure>>,
)>,
known_transactions: LruHashSet<H>,
known_blocks: LruHashSet<B::Hash>,
}
#[derive(Clone, Debug)]
pub struct PeerInfo<B: BlockT> {
pub roles: Roles,
pub best_hash: B::Hash,
pub best_number: <B::Header as HeaderT>::Number,
}
#[derive(Clone)]
pub struct ProtocolConfig {
pub roles: Roles,
pub max_parallel_downloads: u32,
}
impl Default for ProtocolConfig {
fn default() -> ProtocolConfig {
ProtocolConfig {
roles: Roles::FULL,
max_parallel_downloads: 5,
}
}
}
#[derive(Debug, PartialEq, Eq, Clone, Encode, Decode)]
struct BlockAnnouncesHandshake<B: BlockT> {
roles: Roles,
best_number: NumberFor<B>,
best_hash: B::Hash,
genesis_hash: B::Hash,
}
impl<B: BlockT> BlockAnnouncesHandshake<B> {
fn build(
protocol_config: &ProtocolConfig,
best_number: NumberFor<B>,
best_hash: B::Hash,
genesis_hash: B::Hash,
) -> Self {
BlockAnnouncesHandshake {
genesis_hash,
roles: protocol_config.roles,
best_number,
best_hash,
}
}
}
fn build_status_message<B: BlockT>(
protocol_config: &ProtocolConfig,
best_number: NumberFor<B>,
best_hash: B::Hash,
genesis_hash: B::Hash,
) -> Vec<u8> {
let status = message::generic::Status {
version: CURRENT_VERSION,
min_supported_version: MIN_VERSION,
genesis_hash,
roles: protocol_config.roles.into(),
best_number,
best_hash,
chain_status: Vec::new(),
};
Message::<B>::Status(status).encode()
}
impl<B: BlockT, H: ExHashT> Protocol<B, H> {
pub fn new(
config: ProtocolConfig,
chain: Arc<dyn Client<B>>,
transaction_pool: Arc<dyn TransactionPool<H, B>>,
protocol_id: ProtocolId,
config_role: &config::Role,
network_config: &config::NetworkConfiguration,
block_announce_validator: Box<dyn BlockAnnounceValidator<B> + Send>,
metrics_registry: Option<&Registry>,
) -> error::Result<(Protocol<B, H>, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
let info = chain.info();
let sync = ChainSync::new(
config.roles,
chain.clone(),
&info,
block_announce_validator,
config.max_parallel_downloads,
);
let boot_node_ids = {
let mut list = HashSet::new();
for node in &network_config.boot_nodes {
list.insert(node.peer_id.clone());
}
list.shrink_to_fit();
list
};
let important_peers = {
let mut imp_p = HashSet::new();
for reserved in &network_config.default_peers_set.reserved_nodes {
imp_p.insert(reserved.peer_id.clone());
}
for reserved in network_config.extra_sets.iter().flat_map(|s| s.set_config.reserved_nodes.iter()) {
imp_p.insert(reserved.peer_id.clone());
}
imp_p.shrink_to_fit();
imp_p
};
let mut known_addresses = Vec::new();
let (peerset, peerset_handle) = {
let mut sets = Vec::with_capacity(NUM_HARDCODED_PEERSETS + network_config.extra_sets.len());
let mut default_sets_reserved = HashSet::new();
match config_role {
config::Role::Sentry { validators } => {
for validator in validators {
default_sets_reserved.insert(validator.peer_id.clone());
known_addresses.push((validator.peer_id.clone(), validator.multiaddr.clone()));
}
}
config::Role::Authority { sentry_nodes } => {
for sentry_node in sentry_nodes {
default_sets_reserved.insert(sentry_node.peer_id.clone());
known_addresses.push((sentry_node.peer_id.clone(), sentry_node.multiaddr.clone()));
}
}
_ => {}
};
for reserved in network_config.default_peers_set.reserved_nodes.iter() {
default_sets_reserved.insert(reserved.peer_id.clone());
known_addresses.push((reserved.peer_id.clone(), reserved.multiaddr.clone()));
}
let mut bootnodes = Vec::with_capacity(network_config.boot_nodes.len());
for bootnode in network_config.boot_nodes.iter() {
bootnodes.push(bootnode.peer_id.clone());
known_addresses.push((bootnode.peer_id.clone(), bootnode.multiaddr.clone()));
}
sets.push(sc_peerset::SetConfig {
in_peers: network_config.default_peers_set.in_peers,
out_peers: network_config.default_peers_set.out_peers,
bootnodes,
reserved_nodes: default_sets_reserved.clone(),
reserved_only: network_config.default_peers_set.non_reserved_mode
== config::NonReservedPeerMode::Deny,
});
sets.push(sc_peerset::SetConfig {
in_peers: 0,
out_peers: 0,
bootnodes: Vec::new(),
reserved_nodes: HashSet::new(),
reserved_only: true,
});
for set_cfg in &network_config.extra_sets {
let mut reserved_nodes = HashSet::new();
for reserved in set_cfg.set_config.reserved_nodes.iter() {
reserved_nodes.insert(reserved.peer_id.clone());
known_addresses.push((reserved.peer_id.clone(), reserved.multiaddr.clone()));
}
let reserved_only =
set_cfg.set_config.non_reserved_mode == config::NonReservedPeerMode::Deny;
sets.push(sc_peerset::SetConfig {
in_peers: set_cfg.set_config.in_peers,
out_peers: set_cfg.set_config.out_peers,
bootnodes: Vec::new(),
reserved_nodes,
reserved_only,
});
}
sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig {
sets,
})
};
let transactions_protocol: Cow<'static, str> = Cow::from({
let mut proto = String::new();
proto.push_str("/");
proto.push_str(protocol_id.as_ref());
proto.push_str("/transactions/1");
proto
});
let block_announces_protocol: Cow<'static, str> = Cow::from({
let mut proto = String::new();
proto.push_str("/");
proto.push_str(protocol_id.as_ref());
proto.push_str("/block-announces/1");
proto
});
let behaviour = {
let versions = &((MIN_VERSION as u8)..=(CURRENT_VERSION as u8)).collect::<Vec<u8>>();
let handshake_message = Roles::from(config_role).encode();
let best_number = info.best_number;
let best_hash = info.best_hash;
let genesis_hash = info.genesis_hash;
let block_announces_handshake = BlockAnnouncesHandshake::<B>::build(
&config,
best_number,
best_hash,
genesis_hash,
).encode();
GenericProto::new(
protocol_id.clone(),
versions,
build_status_message::<B>(&config, best_number, best_hash, genesis_hash),
peerset,
iter::once((block_announces_protocol, block_announces_handshake, MAX_BLOCK_ANNOUNCE_SIZE))
.chain(iter::once((transactions_protocol, vec![], MAX_TRANSACTIONS_SIZE)))
.chain(network_config.extra_sets.iter().map(|s| (
s.notifications_protocol.clone(),
handshake_message.clone(),
s.max_notification_size
))),
)
};
let block_announce_data_cache = lru::LruCache::new(
network_config.default_peers_set.in_peers as usize
+ network_config.default_peers_set.out_peers as usize,
);
let protocol = Protocol {
tick_timeout: Box::pin(interval(TICK_TIMEOUT)),
propagate_timeout: Box::pin(interval(PROPAGATE_TIMEOUT)),
pending_messages: VecDeque::new(),
pending_transactions: FuturesUnordered::new(),
pending_transactions_peers: HashMap::new(),
config,
peers: HashMap::new(),
chain,
genesis_hash: info.genesis_hash,
sync,
important_peers,
transaction_pool,
peerset_handle: peerset_handle.clone(),
behaviour,
notification_protocols:
network_config.extra_sets.iter().map(|s| s.notifications_protocol.clone()).collect(),
metrics: if let Some(r) = metrics_registry {
Some(Metrics::register(r)?)
} else {
None
},
boot_node_ids,
block_announce_data_cache,
};
Ok((protocol, peerset_handle, known_addresses))
}
pub fn open_peers(&self) -> impl Iterator<Item = &PeerId> {
self.behaviour.open_peers()
}
pub fn requested_peers(&self) -> impl Iterator<Item = &PeerId> {
self.behaviour.requested_peers(HARDCODED_PEERSETS_SYNC)
}
pub fn num_discovered_peers(&self) -> usize {
self.behaviour.num_discovered_peers()
}
pub fn disconnect_peer(&mut self, peer_id: &PeerId, protocol_name: &str) {
if let Some(position) = self.notification_protocols.iter().position(|p| *p == protocol_name) {
self.behaviour.disconnect_peer(peer_id, sc_peerset::SetId::from(position + NUM_HARDCODED_PEERSETS));
} else {
log::warn!(target: "sub-libp2p", "disconnect_peer() with invalid protocol name")
}
}
pub fn peerset_debug_info(&mut self) -> serde_json::Value {
self.behaviour.peerset_debug_info()
}
pub fn num_connected_peers(&self) -> usize {
self.peers.values().count()
}
pub fn num_active_peers(&self) -> usize {
self.peers
.values()
.filter(|p| p.block_request.is_some())
.count()
}
pub fn sync_state(&self) -> SyncState {
self.sync.status().state
}
pub fn best_seen_block(&self) -> Option<NumberFor<B>> {
self.sync.status().best_seen_block
}
pub fn num_sync_peers(&self) -> u32 {
self.sync.status().num_peers
}
pub fn num_queued_blocks(&self) -> u32 {
self.sync.status().queued_blocks
}
pub fn num_downloaded_blocks(&self) -> usize {
self.sync.num_downloaded_blocks()
}
pub fn num_sync_requests(&self) -> usize {
self.sync.num_sync_requests()
}
pub fn new_best_block_imported(&mut self, hash: B::Hash, number: NumberFor<B>) {
trace!(target: "sync", "New best block imported {:?}/#{}", hash, number);
self.sync.update_chain_info(&hash, number);
self.behaviour.set_legacy_handshake_message(
build_status_message::<B>(&self.config, number, hash, self.genesis_hash),
);
self.behaviour.set_notif_protocol_handshake(
HARDCODED_PEERSETS_SYNC,
BlockAnnouncesHandshake::<B>::build(
&self.config,
number,
hash,
self.genesis_hash,
).encode()
);
}
fn update_peer_info(&mut self, who: &PeerId) {
if let Some(info) = self.sync.peer_info(who) {
if let Some(ref mut peer) = self.peers.get_mut(who) {
peer.info.best_hash = info.best_hash;
peer.info.best_number = info.best_number;
}
}
}
pub fn peers_info(&self) -> impl Iterator<Item = (&PeerId, &PeerInfo<B>)> {
self.peers.iter().map(|(id, peer)| (id, &peer.info))
}
fn on_custom_message(
&mut self,
who: PeerId,
data: BytesMut,
) -> CustomMessageOutcome<B> {
let message = match <Message<B> as Decode>::decode(&mut &data[..]) {
Ok(message) => message,
Err(err) => {
debug!(
target: "sync",
"Couldn't decode packet sent by {}: {:?}: {}",
who,
data,
err,
);
self.peerset_handle.report_peer(who, rep::BAD_MESSAGE);
return CustomMessageOutcome::None;
}
};
match message {
GenericMessage::Status(_) =>
debug!(target: "sub-libp2p", "Received unexpected Status"),
GenericMessage::BlockAnnounce(announce) =>
self.push_block_announce_validation(who.clone(), announce),
GenericMessage::Transactions(m) =>
self.on_transactions(who, m),
GenericMessage::BlockResponse(_) =>
warn!(target: "sub-libp2p", "Received unexpected BlockResponse"),
GenericMessage::RemoteCallResponse(_) =>
warn!(target: "sub-libp2p", "Received unexpected RemoteCallResponse"),
GenericMessage::RemoteReadResponse(_) =>
warn!(target: "sub-libp2p", "Received unexpected RemoteReadResponse"),
GenericMessage::RemoteHeaderResponse(_) =>
warn!(target: "sub-libp2p", "Received unexpected RemoteHeaderResponse"),
GenericMessage::RemoteChangesResponse(_) =>
warn!(target: "sub-libp2p", "Received unexpected RemoteChangesResponse"),
GenericMessage::BlockRequest(_) |
GenericMessage::RemoteReadChildRequest(_) |
GenericMessage::RemoteCallRequest(_) |
GenericMessage::RemoteReadRequest(_) |
GenericMessage::RemoteHeaderRequest(_) |
GenericMessage::RemoteChangesRequest(_) |
GenericMessage::Consensus(_) |
GenericMessage::ConsensusBatch(_) => {
debug!(
target: "sub-libp2p",
"Received no longer supported legacy request from {:?}",
who
);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(who, rep::BAD_PROTOCOL);
},
}
CustomMessageOutcome::None
}
fn prepare_block_request(
&mut self,
who: PeerId,
request: message::BlockRequest<B>,
) -> CustomMessageOutcome<B> {
prepare_block_request::<B, H>(&mut self.peers, who, request)
}
pub fn on_sync_peer_disconnected(&mut self, peer: PeerId) -> Result<(), ()> {
if self.important_peers.contains(&peer) {
warn!(target: "sync", "Reserved peer {} disconnected", peer);
} else {
trace!(target: "sync", "{} disconnected", peer);
}
if let Some(_peer_data) = self.peers.remove(&peer) {
self.sync.peer_disconnected(&peer);
Ok(())
} else {
Err(())
}
}
pub fn report_peer(&self, who: PeerId, reputation: sc_peerset::ReputationChange) {
self.peerset_handle.report_peer(who, reputation)
}
pub fn on_block_response(
&mut self,
peer_id: PeerId,
request: message::BlockRequest<B>,
response: crate::schema::v1::BlockResponse,
) -> CustomMessageOutcome<B> {
let blocks = response.blocks.into_iter().map(|block_data| {
Ok(message::BlockData::<B> {
hash: Decode::decode(&mut block_data.hash.as_ref())?,
header: if !block_data.header.is_empty() {
Some(Decode::decode(&mut block_data.header.as_ref())?)
} else {
None
},
body: if request.fields.contains(message::BlockAttributes::BODY) {
Some(block_data.body.iter().map(|body| {
Decode::decode(&mut body.as_ref())
}).collect::<Result<Vec<_>, _>>()?)
} else {
None
},
receipt: if !block_data.message_queue.is_empty() {
Some(block_data.receipt)
} else {
None
},
message_queue: if !block_data.message_queue.is_empty() {
Some(block_data.message_queue)
} else {
None
},
justification: if !block_data.justification.is_empty() {
Some(block_data.justification)
} else if block_data.is_empty_justification {
Some(Vec::new())
} else {
None
},
})
}).collect::<Result<Vec<_>, codec::Error>>();
let blocks = match blocks {
Ok(blocks) => blocks,
Err(err) => {
debug!(target: "sync", "Failed to decode block response from {}: {}", peer_id, err);
self.peerset_handle.report_peer(peer_id, rep::BAD_MESSAGE);
return CustomMessageOutcome::None;
}
};
let block_response = message::BlockResponse::<B> {
id: request.id,
blocks,
};
let blocks_range = || match (
block_response.blocks.first().and_then(|b| b.header.as_ref().map(|h| h.number())),
block_response.blocks.last().and_then(|b| b.header.as_ref().map(|h| h.number())),
) {
(Some(first), Some(last)) if first != last => format!(" ({}..{})", first, last),
(Some(first), Some(_)) => format!(" ({})", first),
_ => Default::default(),
};
trace!(target: "sync", "BlockResponse {} from {} with {} blocks {}",
block_response.id,
peer_id,
block_response.blocks.len(),
blocks_range(),
);
if request.fields == message::BlockAttributes::JUSTIFICATION {
match self.sync.on_block_justification(peer_id, block_response) {
Ok(sync::OnBlockJustification::Nothing) => CustomMessageOutcome::None,
Ok(sync::OnBlockJustification::Import { peer, hash, number, justification }) =>
CustomMessageOutcome::JustificationImport(peer, hash, number, justification),
Err(sync::BadPeer(id, repu)) => {
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(id, repu);
CustomMessageOutcome::None
}
}
} else {
match self.sync.on_block_data(&peer_id, Some(request), block_response) {
Ok(sync::OnBlockData::Import(origin, blocks)) =>
CustomMessageOutcome::BlockImport(origin, blocks),
Ok(sync::OnBlockData::Request(peer, req)) => {
self.prepare_block_request(peer, req)
}
Err(sync::BadPeer(id, repu)) => {
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(id, repu);
CustomMessageOutcome::None
}
}
}
}
pub fn tick(&mut self) {
self.report_metrics()
}
fn on_sync_peer_connected(
&mut self,
who: PeerId,
status: BlockAnnouncesHandshake<B>,
) -> Result<(), ()> {
trace!(target: "sync", "New peer {} {:?}", who, status);
if self.peers.contains_key(&who) {
log::error!(target: "sync", "Called on_sync_peer_connected with already connected peer {}", who);
debug_assert!(false);
return Err(());
}
if status.genesis_hash != self.genesis_hash {
log!(
target: "sync",
if self.important_peers.contains(&who) { Level::Warn } else { Level::Trace },
"Peer is on different chain (our genesis: {} theirs: {})",
self.genesis_hash, status.genesis_hash
);
self.peerset_handle.report_peer(who.clone(), rep::GENESIS_MISMATCH);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
if self.boot_node_ids.contains(&who) {
error!(
target: "sync",
"Bootnode with peer id `{}` is on a different chain (our genesis: {} theirs: {})",
who,
self.genesis_hash,
status.genesis_hash,
);
}
return Err(());
}
if self.config.roles.is_light() {
if status.roles.is_light() {
debug!(target: "sync", "Peer {} is unable to serve light requests", who);
self.peerset_handle.report_peer(who.clone(), rep::BAD_ROLE);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
return Err(());
}
let self_best_block = self
.chain
.info()
.best_number;
let blocks_difference = self_best_block
.checked_sub(&status.best_number)
.unwrap_or_else(Zero::zero)
.saturated_into::<u64>();
if blocks_difference > LIGHT_MAXIMAL_BLOCKS_DIFFERENCE {
debug!(target: "sync", "Peer {} is far behind us and will unable to serve light requests", who);
self.peerset_handle.report_peer(who.clone(), rep::PEER_BEHIND_US_LIGHT);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
return Err(());
}
}
let peer = Peer {
info: PeerInfo {
roles: status.roles,
best_hash: status.best_hash,
best_number: status.best_number
},
block_request: None,
known_transactions: LruHashSet::new(NonZeroUsize::new(MAX_KNOWN_TRANSACTIONS)
.expect("Constant is nonzero")),
known_blocks: LruHashSet::new(NonZeroUsize::new(MAX_KNOWN_BLOCKS)
.expect("Constant is nonzero")),
};
let req = if peer.info.roles.is_full() {
match self.sync.new_peer(who.clone(), peer.info.best_hash, peer.info.best_number) {
Ok(req) => req,
Err(sync::BadPeer(id, repu)) => {
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(id, repu);
return Err(())
}
}
} else {
None
};
debug!(target: "sync", "Connected {}", who);
self.peers.insert(who.clone(), peer);
self.pending_messages.push_back(CustomMessageOutcome::PeerNewBest(who.clone(), status.best_number));
if let Some(req) = req {
let event = self.prepare_block_request(who.clone(), req);
self.pending_messages.push_back(event);
}
Ok(())
}
fn on_transactions(
&mut self,
who: PeerId,
transactions: message::Transactions<B::Extrinsic>,
) {
if !self.config.roles.is_full() {
trace!(target: "sync", "Peer {} is trying to send transactions to the light node", who);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_TX);
self.peerset_handle.report_peer(who, rep::UNEXPECTED_TRANSACTIONS);
return;
}
if self.sync.status().state != SyncState::Idle {
trace!(target: "sync", "{} Ignoring transactions while syncing", who);
return;
}
trace!(target: "sync", "Received {} transactions from {}", transactions.len(), who);
if let Some(ref mut peer) = self.peers.get_mut(&who) {
for t in transactions {
if self.pending_transactions.len() > MAX_PENDING_TRANSACTIONS {
debug!(
target: "sync",
"Ignoring any further transactions that exceed `MAX_PENDING_TRANSACTIONS`({}) limit",
MAX_PENDING_TRANSACTIONS,
);
break;
}
let hash = self.transaction_pool.hash_of(&t);
peer.known_transactions.insert(hash.clone());
self.peerset_handle.report_peer(who.clone(), rep::ANY_TRANSACTION);
match self.pending_transactions_peers.entry(hash.clone()) {
Entry::Vacant(entry) => {
self.pending_transactions.push(PendingTransaction {
validation: self.transaction_pool.import(t),
tx_hash: hash,
});
entry.insert(vec![who.clone()]);
},
Entry::Occupied(mut entry) => {
entry.get_mut().push(who.clone());
}
}
}
}
}
fn on_handle_transaction_import(&mut self, who: PeerId, import: TransactionImport) {
match import {
TransactionImport::KnownGood => self.peerset_handle.report_peer(who, rep::ANY_TRANSACTION_REFUND),
TransactionImport::NewGood => self.peerset_handle.report_peer(who, rep::GOOD_TRANSACTION),
TransactionImport::Bad => self.peerset_handle.report_peer(who, rep::BAD_TRANSACTION),
TransactionImport::None => {},
}
}
pub fn propagate_transaction(
&mut self,
hash: &H,
) {
debug!(target: "sync", "Propagating transaction [{:?}]", hash);
if self.sync.status().state != SyncState::Idle {
return;
}
if let Some(transaction) = self.transaction_pool.transaction(hash) {
let propagated_to = self.do_propagate_transactions(&[(hash.clone(), transaction)]);
self.transaction_pool.on_broadcasted(propagated_to);
}
}
fn do_propagate_transactions(
&mut self,
transactions: &[(H, B::Extrinsic)],
) -> HashMap<H, Vec<String>> {
let mut propagated_to = HashMap::<_, Vec<_>>::new();
let mut propagated_transactions = 0;
for (who, peer) in self.peers.iter_mut() {
if !peer.info.roles.is_full() {
continue;
}
if !self.behaviour.is_open(who, HARDCODED_PEERSETS_TX) {
continue;
}
let (hashes, to_send): (Vec<_>, Vec<_>) = transactions
.iter()
.filter(|&(ref hash, _)| peer.known_transactions.insert(hash.clone()))
.cloned()
.unzip();
propagated_transactions += hashes.len();
if !to_send.is_empty() {
for hash in hashes {
propagated_to
.entry(hash)
.or_default()
.push(who.to_base58());
}
trace!(target: "sync", "Sending {} transactions to {}", to_send.len(), who);
self.behaviour.write_notification(
who,
HARDCODED_PEERSETS_TX,
to_send.encode()
);
}
}
if let Some(ref metrics) = self.metrics {
metrics.propagated_transactions.inc_by(propagated_transactions as _)
}
propagated_to
}
pub fn propagate_transactions(&mut self) {
debug!(target: "sync", "Propagating transactions");
if self.sync.status().state != SyncState::Idle {
return;
}
let transactions = self.transaction_pool.transactions();
let propagated_to = self.do_propagate_transactions(&transactions);
self.transaction_pool.on_broadcasted(propagated_to);
}
pub fn announce_block(&mut self, hash: B::Hash, data: Option<Vec<u8>>) {
let header = match self.chain.header(BlockId::Hash(hash)) {
Ok(Some(header)) => header,
Ok(None) => {
warn!("Trying to announce unknown block: {}", hash);
return;
}
Err(e) => {
warn!("Error reading block header {}: {:?}", hash, e);
return;
}
};
if header.number().is_zero() {
return;
}
let is_best = self.chain.info().best_hash == hash;
debug!(target: "sync", "Reannouncing block {:?} is_best: {}", hash, is_best);
let data = data.or_else(|| self.block_announce_data_cache.get(&hash).cloned()).unwrap_or_default();
for (who, ref mut peer) in self.peers.iter_mut() {
let inserted = peer.known_blocks.insert(hash);
if inserted {
trace!(target: "sync", "Announcing block {:?} to {}", hash, who);
let message = message::BlockAnnounce {
header: header.clone(),
state: if is_best {
Some(message::BlockState::Best)
} else {
Some(message::BlockState::Normal)
},
data: Some(data.clone()),
};
self.behaviour.write_notification(
who,
HARDCODED_PEERSETS_SYNC,
message.encode()
);
}
}
}
fn push_block_announce_validation(
&mut self,
who: PeerId,
announce: BlockAnnounce<B::Header>,
) {
let hash = announce.header.hash();
let peer = match self.peers.get_mut(&who) {
Some(p) => p,
None => {
log::error!(target: "sync", "Received block announce from disconnected peer {}", who);
debug_assert!(false);
return;
}
};
peer.known_blocks.insert(hash.clone());
let is_best = match announce.state.unwrap_or(message::BlockState::Best) {
message::BlockState::Best => true,
message::BlockState::Normal => false,
};
if peer.info.roles.is_full() {
self.sync.push_block_announce_validation(who, hash, announce, is_best);
}
}
fn process_block_announce_validation_result(
&mut self,
validation_result: sync::PollBlockAnnounceValidation<B::Header>,
) -> CustomMessageOutcome<B> {
let (header, is_best, who) = match validation_result {
sync::PollBlockAnnounceValidation::Skip =>
return CustomMessageOutcome::None,
sync::PollBlockAnnounceValidation::Nothing { is_best, who, announce } => {
self.update_peer_info(&who);
if let Some(data) = announce.data {
if !data.is_empty() {
self.block_announce_data_cache.put(announce.header.hash(), data);
}
}
if is_best {
return CustomMessageOutcome::PeerNewBest(who, *announce.header.number())
} else {
return CustomMessageOutcome::None
}
}
sync::PollBlockAnnounceValidation::ImportHeader { announce, is_best, who } => {
self.update_peer_info(&who);
if let Some(data) = announce.data {
if !data.is_empty() {
self.block_announce_data_cache.put(announce.header.hash(), data);
}
}
(announce.header, is_best, who)
}
sync::PollBlockAnnounceValidation::Failure { who, disconnect } => {
if disconnect {
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
}
self.report_peer(who, rep::BAD_BLOCK_ANNOUNCEMENT);
return CustomMessageOutcome::None
}
};
let number = *header.number();
let blocks_to_import = self.sync.on_block_data(
&who,
None,
message::generic::BlockResponse {
id: 0,
blocks: vec![
message::generic::BlockData {
hash: header.hash(),
header: Some(header),
body: None,
receipt: None,
message_queue: None,
justification: None,
},
],
},
);
if is_best {
self.pending_messages.push_back(
CustomMessageOutcome::PeerNewBest(who, number),
);
}
match blocks_to_import {
Ok(sync::OnBlockData::Import(origin, blocks)) => {
CustomMessageOutcome::BlockImport(origin, blocks)
},
Ok(sync::OnBlockData::Request(peer, req)) => {
self.prepare_block_request(peer, req)
}
Err(sync::BadPeer(id, repu)) => {
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(id, repu);
CustomMessageOutcome::None
}
}
}
pub fn on_block_finalized(&mut self, hash: B::Hash, header: &B::Header) {
self.sync.on_block_finalized(&hash, *header.number())
}
pub fn request_justification(&mut self, hash: &B::Hash, number: NumberFor<B>) {
self.sync.request_justification(&hash, number)
}
pub fn set_sync_fork_request(&mut self, peers: Vec<PeerId>, hash: &B::Hash, number: NumberFor<B>) {
self.sync.set_sync_fork_request(peers, hash, number)
}
pub fn on_blocks_processed(
&mut self,
imported: usize,
count: usize,
results: Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>
) {
let results = self.sync.on_blocks_processed(
imported,
count,
results,
);
for result in results {
match result {
Ok((id, req)) => {
self.pending_messages.push_back(
prepare_block_request(&mut self.peers, id, req)
);
}
Err(sync::BadPeer(id, repu)) => {
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(id, repu)
}
}
}
}
pub fn justification_import_result(&mut self, who: PeerId, hash: B::Hash, number: NumberFor<B>, success: bool) {
self.sync.on_justification_import(hash, number, success);
if !success {
log::info!("💔 Invalid justification provided by {} for #{}", who, hash);
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
self.peerset_handle.report_peer(
who,
sc_peerset::ReputationChange::new_fatal("Invalid justification")
);
}
}
pub fn set_reserved_only(&self, reserved_only: bool) {
self.peerset_handle.set_reserved_only(HARDCODED_PEERSETS_SYNC, reserved_only);
self.peerset_handle.set_reserved_only(HARDCODED_PEERSETS_TX, reserved_only);
}
pub fn remove_reserved_peer(&self, peer: PeerId) {
self.peerset_handle.remove_reserved_peer(HARDCODED_PEERSETS_SYNC, peer.clone());
self.peerset_handle.remove_reserved_peer(HARDCODED_PEERSETS_TX, peer);
}
pub fn add_reserved_peer(&self, peer: PeerId) {
self.peerset_handle.add_reserved_peer(HARDCODED_PEERSETS_SYNC, peer.clone());
self.peerset_handle.add_reserved_peer(HARDCODED_PEERSETS_TX, peer);
}
pub fn set_reserved_peers(&self, peers: HashSet<PeerId>) {
self.peerset_handle.set_reserved_peers(HARDCODED_PEERSETS_SYNC, peers.clone());
self.peerset_handle.set_reserved_peers(HARDCODED_PEERSETS_TX, peers);
}
pub fn remove_set_reserved_peer(&self, protocol: Cow<'static, str>, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.remove_reserved_peer(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
} else {
log::error!(
target: "sub-libp2p",
"remove_set_reserved_peer with unknown protocol: {}",
protocol
);
}
}
pub fn add_set_reserved_peer(&self, protocol: Cow<'static, str>, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.add_reserved_peer(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
} else {
log::error!(
target: "sub-libp2p",
"add_set_reserved_peer with unknown protocol: {}",
protocol
);
}
}
pub fn add_default_set_discovered_nodes(&mut self, peer_ids: impl Iterator<Item = PeerId>) {
for peer_id in peer_ids {
self.peerset_handle.add_to_peers_set(HARDCODED_PEERSETS_SYNC, peer_id);
}
}
pub fn add_to_peers_set(&self, protocol: Cow<'static, str>, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.add_to_peers_set(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
} else {
log::error!(
target: "sub-libp2p",
"add_to_peers_set with unknown protocol: {}",
protocol
);
}
}
pub fn remove_from_peers_set(&self, protocol: Cow<'static, str>, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.remove_from_peers_set(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
} else {
log::error!(
target: "sub-libp2p",
"remove_from_peers_set with unknown protocol: {}",
protocol
);
}
}
fn report_metrics(&self) {
if let Some(metrics) = &self.metrics {
let n = u64::try_from(self.peers.len()).unwrap_or(std::u64::MAX);
metrics.peers.set(n);
let m = self.sync.metrics();
metrics.fork_targets.set(m.fork_targets.into());
metrics.queued_blocks.set(m.queued_blocks.into());
metrics.justifications.with_label_values(&["pending"])
.set(m.justifications.pending_requests.into());
metrics.justifications.with_label_values(&["active"])
.set(m.justifications.active_requests.into());
metrics.justifications.with_label_values(&["failed"])
.set(m.justifications.failed_requests.into());
metrics.justifications.with_label_values(&["importing"])
.set(m.justifications.importing_requests.into());
}
}
}
fn prepare_block_request<B: BlockT, H: ExHashT>(
peers: &mut HashMap<PeerId, Peer<B, H>>,
who: PeerId,
request: message::BlockRequest<B>,
) -> CustomMessageOutcome<B> {
let (tx, rx) = oneshot::channel();
if let Some(ref mut peer) = peers.get_mut(&who) {
peer.block_request = Some((request.clone(), rx));
}
let request = crate::schema::v1::BlockRequest {
fields: request.fields.to_be_u32(),
from_block: match request.from {
message::FromBlock::Hash(h) =>
Some(crate::schema::v1::block_request::FromBlock::Hash(h.encode())),
message::FromBlock::Number(n) =>
Some(crate::schema::v1::block_request::FromBlock::Number(n.encode())),
},
to_block: request.to.map(|h| h.encode()).unwrap_or_default(),
direction: request.direction as i32,
max_blocks: request.max.unwrap_or(0),
};
CustomMessageOutcome::BlockRequest {
target: who,
request: request,
pending_response: tx,
}
}
#[derive(Debug)]
#[must_use]
pub enum CustomMessageOutcome<B: BlockT> {
BlockImport(BlockOrigin, Vec<IncomingBlock<B>>),
JustificationImport(Origin, B::Hash, NumberFor<B>, Justification),
NotificationStreamOpened {
remote: PeerId,
protocol: Cow<'static, str>,
roles: Roles,
notifications_sink: NotificationsSink
},
NotificationStreamReplaced {
remote: PeerId,
protocol: Cow<'static, str>,
notifications_sink: NotificationsSink,
},
NotificationStreamClosed { remote: PeerId, protocol: Cow<'static, str> },
NotificationsReceived { remote: PeerId, messages: Vec<(Cow<'static, str>, Bytes)> },
BlockRequest {
target: PeerId,
request: crate::schema::v1::BlockRequest,
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
},
PeerNewBest(PeerId, NumberFor<B>),
SyncConnected(PeerId),
SyncDisconnected(PeerId),
None,
}
impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
type ProtocolsHandler = <GenericProto as NetworkBehaviour>::ProtocolsHandler;
type OutEvent = CustomMessageOutcome<B>;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
self.behaviour.new_handler()
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
self.behaviour.addresses_of_peer(peer_id)
}
fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
self.behaviour.inject_connection_established(peer_id, conn, endpoint)
}
fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
self.behaviour.inject_connection_closed(peer_id, conn, endpoint)
}
fn inject_connected(&mut self, peer_id: &PeerId) {
self.behaviour.inject_connected(peer_id)
}
fn inject_disconnected(&mut self, peer_id: &PeerId) {
self.behaviour.inject_disconnected(peer_id)
}
fn inject_event(
&mut self,
peer_id: PeerId,
connection: ConnectionId,
event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent,
) {
self.behaviour.inject_event(peer_id, connection, event)
}
fn poll(
&mut self,
cx: &mut std::task::Context,
params: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent,
Self::OutEvent
>
> {
if let Some(message) = self.pending_messages.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(message));
}
let mut finished_block_requests = Vec::new();
for (id, peer) in self.peers.iter_mut() {
if let Peer { block_request: Some((_, pending_response)), .. } = peer {
match pending_response.poll_unpin(cx) {
Poll::Ready(Ok(Ok(resp))) => {
let (req, _) = peer.block_request.take().unwrap();
let protobuf_response = match crate::schema::v1::BlockResponse::decode(&resp[..]) {
Ok(proto) => proto,
Err(e) => {
trace!(target: "sync", "Failed to decode block request to peer {:?}: {:?}.", id, e);
self.peerset_handle.report_peer(id.clone(), rep::BAD_MESSAGE);
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
continue;
}
};
finished_block_requests.push((id.clone(), req, protobuf_response));
},
Poll::Ready(Ok(Err(e))) => {
peer.block_request.take();
trace!(target: "sync", "Block request to peer {:?} failed: {:?}.", id, e);
match e {
RequestFailure::Network(OutboundFailure::Timeout) => {
self.peerset_handle.report_peer(id.clone(), rep::TIMEOUT);
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
}
RequestFailure::Network(OutboundFailure::UnsupportedProtocols) => {
self.peerset_handle.report_peer(id.clone(), rep::BAD_PROTOCOL);
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
}
RequestFailure::Network(OutboundFailure::DialFailure) => {
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
}
RequestFailure::Refused => {
self.peerset_handle.report_peer(id.clone(), rep::REFUSED);
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
}
RequestFailure::Network(OutboundFailure::ConnectionClosed)
| RequestFailure::NotConnected => {
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
},
RequestFailure::UnknownProtocol => {
debug_assert!(false, "Block request protocol should always be known.");
}
RequestFailure::Obsolete => {
debug_assert!(
false,
"Can not receive `RequestFailure::Obsolete` after dropping the \
response receiver.",
);
}
}
},
Poll::Ready(Err(oneshot::Canceled)) => {
peer.block_request.take();
trace!(
target: "sync",
"Block request to peer {:?} failed due to oneshot being canceled.",
id,
);
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
},
Poll::Pending => {},
}
}
}
for (id, req, protobuf_response) in finished_block_requests {
let ev = self.on_block_response(id, req, protobuf_response);
self.pending_messages.push_back(ev);
}
while let Poll::Ready(Some(())) = self.tick_timeout.poll_next_unpin(cx) {
self.tick();
}
while let Poll::Ready(Some(())) = self.propagate_timeout.poll_next_unpin(cx) {
self.propagate_transactions();
}
for (id, request) in self.sync.block_requests() {
let event = prepare_block_request(&mut self.peers, id.clone(), request);
self.pending_messages.push_back(event);
}
for (id, request) in self.sync.justification_requests() {
let event = prepare_block_request(&mut self.peers, id, request);
self.pending_messages.push_back(event);
}
if let Poll::Ready(Some((tx_hash, result))) = self.pending_transactions.poll_next_unpin(cx) {
if let Some(peers) = self.pending_transactions_peers.remove(&tx_hash) {
peers.into_iter().for_each(|p| self.on_handle_transaction_import(p, result));
} else {
warn!(target: "sub-libp2p", "Inconsistent state, no peers for pending transaction!");
}
}
while let Poll::Ready(result) = self.sync.poll_block_announce_validation(cx) {
match self.process_block_announce_validation_result(result) {
CustomMessageOutcome::None => {},
outcome => self.pending_messages.push_back(outcome),
}
}
if let Some(message) = self.pending_messages.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(message));
}
let event = match self.behaviour.poll(cx, params) {
Poll::Pending => return Poll::Pending,
Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev)) => ev,
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }) =>
return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) =>
return Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }),
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address, score }) =>
return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address, score }),
};
let outcome = match event {
GenericProtoOut::CustomProtocolOpen { peer_id, set_id, received_handshake, notifications_sink, .. } => {
if set_id == HARDCODED_PEERSETS_SYNC {
match <Message<B> as DecodeAll>::decode_all(&mut &received_handshake[..]) {
Ok(GenericMessage::Status(handshake)) => {
let handshake = BlockAnnouncesHandshake {
roles: handshake.roles,
best_number: handshake.best_number,
best_hash: handshake.best_hash,
genesis_hash: handshake.genesis_hash,
};
if self.on_sync_peer_connected(peer_id.clone(), handshake).is_ok() {
self.peerset_handle.add_reserved_peer(
HARDCODED_PEERSETS_TX,
peer_id.clone()
);
CustomMessageOutcome::SyncConnected(peer_id)
} else {
CustomMessageOutcome::None
}
},
Ok(msg) => {
debug!(
target: "sync",
"Expected Status message from {}, but got {:?}",
peer_id,
msg,
);
self.peerset_handle.report_peer(peer_id, rep::BAD_MESSAGE);
CustomMessageOutcome::None
}
Err(err) => {
match <BlockAnnouncesHandshake<B> as DecodeAll>::decode_all(&mut &received_handshake[..]) {
Ok(handshake) => {
if self.on_sync_peer_connected(peer_id.clone(), handshake).is_ok() {
self.peerset_handle.add_reserved_peer(
HARDCODED_PEERSETS_TX,
peer_id.clone()
);
CustomMessageOutcome::SyncConnected(peer_id)
} else {
CustomMessageOutcome::None
}
}
Err(err2) => {
debug!(
target: "sync",
"Couldn't decode handshake sent by {}: {:?}: {} & {}",
peer_id,
received_handshake,
err,
err2,
);
self.peerset_handle.report_peer(peer_id, rep::BAD_MESSAGE);
CustomMessageOutcome::None
}
}
}
}
} else if set_id == HARDCODED_PEERSETS_TX {
CustomMessageOutcome::None
} else {
match message::Roles::decode_all(&received_handshake[..]) {
Ok(roles) =>
CustomMessageOutcome::NotificationStreamOpened {
remote: peer_id,
protocol: self.notification_protocols[usize::from(set_id) - NUM_HARDCODED_PEERSETS].clone(),
roles,
notifications_sink,
},
Err(err) => {
debug!(target: "sync", "Failed to parse remote handshake: {}", err);
self.behaviour.disconnect_peer(&peer_id, set_id);
self.peerset_handle.report_peer(peer_id, rep::BAD_MESSAGE);
CustomMessageOutcome::None
}
}
}
}
GenericProtoOut::CustomProtocolReplaced { peer_id, notifications_sink, set_id } => {
if set_id == HARDCODED_PEERSETS_SYNC || set_id == HARDCODED_PEERSETS_TX {
CustomMessageOutcome::None
} else {
CustomMessageOutcome::NotificationStreamReplaced {
remote: peer_id,
protocol: self.notification_protocols[usize::from(set_id) - NUM_HARDCODED_PEERSETS].clone(),
notifications_sink,
}
}
},
GenericProtoOut::CustomProtocolClosed { peer_id, set_id } => {
if set_id == HARDCODED_PEERSETS_SYNC {
if self.on_sync_peer_disconnected(peer_id.clone()).is_ok() {
self.peerset_handle.remove_reserved_peer(
HARDCODED_PEERSETS_TX,
peer_id.clone()
);
CustomMessageOutcome::SyncDisconnected(peer_id)
} else {
log::debug!(
target: "sync",
"Disconnected peer which had earlier been refused by on_sync_peer_connected {}",
peer_id
);
CustomMessageOutcome::None
}
} else if set_id == HARDCODED_PEERSETS_TX {
CustomMessageOutcome::None
} else {
CustomMessageOutcome::NotificationStreamClosed {
remote: peer_id,
protocol: self.notification_protocols[usize::from(set_id) - NUM_HARDCODED_PEERSETS].clone(),
}
}
},
GenericProtoOut::LegacyMessage { peer_id, message } => {
if self.peers.contains_key(&peer_id) {
self.on_custom_message(peer_id, message)
} else {
CustomMessageOutcome::None
}
},
GenericProtoOut::Notification { peer_id, set_id, message } =>
match set_id {
HARDCODED_PEERSETS_SYNC if self.peers.contains_key(&peer_id) => {
if let Ok(announce) = message::BlockAnnounce::decode(&mut message.as_ref()) {
self.push_block_announce_validation(peer_id, announce);
if let Poll::Ready(res) = self.sync.poll_block_announce_validation(cx) {
self.process_block_announce_validation_result(res)
} else {
CustomMessageOutcome::None
}
} else {
warn!(target: "sub-libp2p", "Failed to decode block announce");
CustomMessageOutcome::None
}
}
HARDCODED_PEERSETS_TX if self.peers.contains_key(&peer_id) => {
if let Ok(m) = <message::Transactions<B::Extrinsic> as Decode>::decode(
&mut message.as_ref(),
) {
self.on_transactions(peer_id, m);
} else {
warn!(target: "sub-libp2p", "Failed to decode transactions list");
}
CustomMessageOutcome::None
}
HARDCODED_PEERSETS_SYNC | HARDCODED_PEERSETS_TX => {
debug!(
target: "sync",
"Received sync or transaction for peer earlier refused by sync layer: {}",
peer_id
);
CustomMessageOutcome::None
}
_ => {
let protocol_name = self.notification_protocols[usize::from(set_id) - NUM_HARDCODED_PEERSETS].clone();
CustomMessageOutcome::NotificationsReceived {
remote: peer_id,
messages: vec![(protocol_name, message.freeze())],
}
}
}
};
if !matches!(outcome, CustomMessageOutcome::<B>::None) {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(outcome));
}
if let Some(message) = self.pending_messages.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(message));
}
cx.waker().wake_by_ref();
Poll::Pending
}
fn inject_addr_reach_failure(
&mut self,
peer_id: Option<&PeerId>,
addr: &Multiaddr,
error: &dyn std::error::Error
) {
self.behaviour.inject_addr_reach_failure(peer_id, addr, error)
}
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
self.behaviour.inject_dial_failure(peer_id)
}
fn inject_new_listen_addr(&mut self, addr: &Multiaddr) {
self.behaviour.inject_new_listen_addr(addr)
}
fn inject_expired_listen_addr(&mut self, addr: &Multiaddr) {
self.behaviour.inject_expired_listen_addr(addr)
}
fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
self.behaviour.inject_new_external_addr(addr)
}
fn inject_listener_error(&mut self, id: ListenerId, err: &(dyn std::error::Error + 'static)) {
self.behaviour.inject_listener_error(id, err);
}
fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
self.behaviour.inject_listener_closed(id, reason);
}
}