mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-06-12 10:01:17 +00:00
Clean up sc-network (#9761)
* Clean up sc-network - Avoid using clone() for the Copy type `PeerId`. - Use `find_map` for `filter_map` and `next`. - Use `Self`. * More on Copy types * Cargo +nightly fmt --all * More .. * fmt * Revert vec![default_notif_handshake_message]
This commit is contained in:
@@ -56,7 +56,7 @@ fn get_addresses_and_authority_id() {
|
||||
let remote_addr = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
.with(Protocol::P2p(remote_peer_id.clone().into()));
|
||||
.with(Protocol::P2p(remote_peer_id.into()));
|
||||
|
||||
let test_api = Arc::new(TestApi { authorities: vec![] });
|
||||
|
||||
|
||||
@@ -289,7 +289,7 @@ where
|
||||
if a.iter().any(|p| matches!(p, multiaddr::Protocol::P2p(_))) {
|
||||
a
|
||||
} else {
|
||||
a.with(multiaddr::Protocol::P2p(peer_id.clone()))
|
||||
a.with(multiaddr::Protocol::P2p(peer_id))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -199,7 +199,7 @@ mod tests {
|
||||
let multiaddr1 = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
.with(Protocol::P2p(peer_id.clone().into()));
|
||||
.with(Protocol::P2p(peer_id.into()));
|
||||
let multiaddr2 = "/ip6/2002:db8:0:0:0:0:0:2/tcp/30133"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
|
||||
@@ -167,7 +167,7 @@ impl NetworkProvider for TestNetwork {
|
||||
|
||||
impl NetworkStateInfo for TestNetwork {
|
||||
fn local_peer_id(&self) -> PeerId {
|
||||
self.peer_id.clone()
|
||||
self.peer_id
|
||||
}
|
||||
|
||||
fn external_addresses(&self) -> Vec<Multiaddr> {
|
||||
|
||||
@@ -593,7 +593,7 @@ impl<N: Ord> Peers<N> {
|
||||
let mut peers = self
|
||||
.inner
|
||||
.iter()
|
||||
.map(|(peer_id, info)| (peer_id.clone(), info.clone()))
|
||||
.map(|(peer_id, info)| (*peer_id, info.clone()))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
peers.shuffle(&mut rand::thread_rng());
|
||||
@@ -618,9 +618,9 @@ impl<N: Ord> Peers<N> {
|
||||
let mut n_authorities_added = 0;
|
||||
for peer_id in shuffled_authorities {
|
||||
if n_authorities_added < half_lucky {
|
||||
first_stage_peers.insert(peer_id.clone());
|
||||
first_stage_peers.insert(*peer_id);
|
||||
} else if n_authorities_added < one_and_a_half_lucky {
|
||||
second_stage_peers.insert(peer_id.clone());
|
||||
second_stage_peers.insert(*peer_id);
|
||||
} else {
|
||||
break
|
||||
}
|
||||
@@ -637,11 +637,11 @@ impl<N: Ord> Peers<N> {
|
||||
}
|
||||
|
||||
if first_stage_peers.len() < LUCKY_PEERS {
|
||||
first_stage_peers.insert(peer_id.clone());
|
||||
first_stage_peers.insert(*peer_id);
|
||||
second_stage_peers.remove(peer_id);
|
||||
} else if second_stage_peers.len() < n_second_stage_peers {
|
||||
if !first_stage_peers.contains(peer_id) {
|
||||
second_stage_peers.insert(peer_id.clone());
|
||||
second_stage_peers.insert(*peer_id);
|
||||
}
|
||||
} else {
|
||||
break
|
||||
|
||||
@@ -696,10 +696,10 @@ mod tests {
|
||||
let mut network = NoOpNetwork::default();
|
||||
|
||||
let peer_id = PeerId::random();
|
||||
consensus.new_peer(&mut network, peer_id.clone(), ObservedRole::Full);
|
||||
consensus.new_peer(&mut network, peer_id, ObservedRole::Full);
|
||||
assert!(consensus.peers.contains_key(&peer_id));
|
||||
|
||||
consensus.peer_disconnected(&mut network, peer_id.clone());
|
||||
consensus.peer_disconnected(&mut network, peer_id);
|
||||
assert!(!consensus.peers.contains_key(&peer_id));
|
||||
}
|
||||
|
||||
|
||||
@@ -222,7 +222,7 @@ impl<B: BlockT> Behaviour<B> {
|
||||
request_response_protocols.push(state_request_protocol_config);
|
||||
request_response_protocols.push(light_client_request_protocol_config);
|
||||
|
||||
Ok(Behaviour {
|
||||
Ok(Self {
|
||||
substrate,
|
||||
peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
|
||||
discovery: disco_config.finish(),
|
||||
|
||||
@@ -190,7 +190,7 @@ pub struct Bitswap<B> {
|
||||
impl<B: BlockT> Bitswap<B> {
|
||||
/// Create a new instance of the bitswap protocol handler.
|
||||
pub fn new(client: Arc<dyn Client<B>>) -> Self {
|
||||
Bitswap { client, ready_blocks: Default::default() }
|
||||
Self { client, ready_blocks: Default::default() }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -305,7 +305,7 @@ impl<B: BlockT> NetworkBehaviour for Bitswap<B> {
|
||||
>{
|
||||
if let Some((peer_id, message)) = self.ready_blocks.pop_front() {
|
||||
return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id,
|
||||
handler: NotifyHandler::Any,
|
||||
event: message,
|
||||
})
|
||||
|
||||
@@ -169,9 +169,9 @@ impl Role {
|
||||
impl fmt::Display for Role {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Role::Full => write!(f, "FULL"),
|
||||
Role::Light => write!(f, "LIGHT"),
|
||||
Role::Authority { .. } => write!(f, "AUTHORITY"),
|
||||
Self::Full => write!(f, "FULL"),
|
||||
Self::Light => write!(f, "LIGHT"),
|
||||
Self::Authority { .. } => write!(f, "AUTHORITY"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -242,7 +242,7 @@ pub struct ProtocolId(smallvec::SmallVec<[u8; 6]>);
|
||||
|
||||
impl<'a> From<&'a str> for ProtocolId {
|
||||
fn from(bytes: &'a str) -> ProtocolId {
|
||||
ProtocolId(bytes.as_bytes().into())
|
||||
Self(bytes.as_bytes().into())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -313,7 +313,7 @@ pub struct MultiaddrWithPeerId {
|
||||
impl MultiaddrWithPeerId {
|
||||
/// Concatenates the multiaddress and peer ID into one multiaddress containing both.
|
||||
pub fn concat(&self) -> Multiaddr {
|
||||
let proto = multiaddr::Protocol::P2p(From::from(self.peer_id.clone()));
|
||||
let proto = multiaddr::Protocol::P2p(From::from(self.peer_id));
|
||||
self.multiaddr.clone().with(proto)
|
||||
}
|
||||
}
|
||||
@@ -360,9 +360,9 @@ pub enum ParseErr {
|
||||
impl fmt::Display for ParseErr {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
ParseErr::MultiaddrParse(err) => write!(f, "{}", err),
|
||||
ParseErr::InvalidPeerId => write!(f, "Peer id at the end of the address is invalid"),
|
||||
ParseErr::PeerIdMissing => write!(f, "Peer id is missing from the address"),
|
||||
Self::MultiaddrParse(err) => write!(f, "{}", err),
|
||||
Self::InvalidPeerId => write!(f, "Peer id at the end of the address is invalid"),
|
||||
Self::PeerIdMissing => write!(f, "Peer id is missing from the address"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -370,16 +370,16 @@ impl fmt::Display for ParseErr {
|
||||
impl std::error::Error for ParseErr {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
ParseErr::MultiaddrParse(err) => Some(err),
|
||||
ParseErr::InvalidPeerId => None,
|
||||
ParseErr::PeerIdMissing => None,
|
||||
Self::MultiaddrParse(err) => Some(err),
|
||||
Self::InvalidPeerId => None,
|
||||
Self::PeerIdMissing => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<multiaddr::Error> for ParseErr {
|
||||
fn from(err: multiaddr::Error) -> ParseErr {
|
||||
ParseErr::MultiaddrParse(err)
|
||||
Self::MultiaddrParse(err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -401,7 +401,7 @@ pub enum SyncMode {
|
||||
|
||||
impl Default for SyncMode {
|
||||
fn default() -> Self {
|
||||
SyncMode::Full
|
||||
Self::Full
|
||||
}
|
||||
}
|
||||
|
||||
@@ -479,7 +479,7 @@ impl NetworkConfiguration {
|
||||
node_key: NodeKeyConfig,
|
||||
net_config_path: Option<PathBuf>,
|
||||
) -> Self {
|
||||
NetworkConfiguration {
|
||||
Self {
|
||||
net_config_path,
|
||||
listen_addresses: Vec::new(),
|
||||
public_addresses: Vec::new(),
|
||||
@@ -548,7 +548,7 @@ pub struct SetConfig {
|
||||
|
||||
impl Default for SetConfig {
|
||||
fn default() -> Self {
|
||||
SetConfig {
|
||||
Self {
|
||||
in_peers: 25,
|
||||
out_peers: 75,
|
||||
reserved_nodes: Vec::new(),
|
||||
@@ -585,7 +585,7 @@ pub struct NonDefaultSetConfig {
|
||||
impl NonDefaultSetConfig {
|
||||
/// Creates a new [`NonDefaultSetConfig`]. Zero slots and accepts only reserved nodes.
|
||||
pub fn new(notifications_protocol: Cow<'static, str>, max_notification_size: u64) -> Self {
|
||||
NonDefaultSetConfig {
|
||||
Self {
|
||||
notifications_protocol,
|
||||
max_notification_size,
|
||||
fallback_names: Vec::new(),
|
||||
@@ -644,8 +644,8 @@ impl NonReservedPeerMode {
|
||||
/// Attempt to parse the peer mode from a string.
|
||||
pub fn parse(s: &str) -> Option<Self> {
|
||||
match s {
|
||||
"accept" => Some(NonReservedPeerMode::Accept),
|
||||
"deny" => Some(NonReservedPeerMode::Deny),
|
||||
"accept" => Some(Self::Accept),
|
||||
"deny" => Some(Self::Deny),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -662,7 +662,7 @@ pub enum NodeKeyConfig {
|
||||
|
||||
impl Default for NodeKeyConfig {
|
||||
fn default() -> NodeKeyConfig {
|
||||
NodeKeyConfig::Ed25519(Secret::New)
|
||||
Self::Ed25519(Secret::New)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -687,9 +687,9 @@ pub enum Secret<K> {
|
||||
impl<K> fmt::Debug for Secret<K> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
match self {
|
||||
Secret::Input(_) => f.debug_tuple("Secret::Input").finish(),
|
||||
Secret::File(path) => f.debug_tuple("Secret::File").field(path).finish(),
|
||||
Secret::New => f.debug_tuple("Secret::New").finish(),
|
||||
Self::Input(_) => f.debug_tuple("Secret::Input").finish(),
|
||||
Self::File(path) => f.debug_tuple("Secret::File").field(path).finish(),
|
||||
Self::New => f.debug_tuple("Secret::New").finish(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,7 +71,7 @@ use libp2p::{
|
||||
NetworkBehaviourAction, PollParameters, ProtocolsHandler,
|
||||
},
|
||||
};
|
||||
use log::{debug, info, trace, warn};
|
||||
use log::{debug, error, info, trace, warn};
|
||||
use sp_core::hexdisplay::HexDisplay;
|
||||
use std::{
|
||||
cmp,
|
||||
@@ -106,7 +106,7 @@ pub struct DiscoveryConfig {
|
||||
impl DiscoveryConfig {
|
||||
/// Create a default configuration with the given public key.
|
||||
pub fn new(local_public_key: PublicKey) -> Self {
|
||||
DiscoveryConfig {
|
||||
Self {
|
||||
local_peer_id: local_public_key.into_peer_id(),
|
||||
permanent_addresses: Vec::new(),
|
||||
dht_random_walk: true,
|
||||
@@ -180,7 +180,7 @@ impl DiscoveryConfig {
|
||||
|
||||
/// Create a `DiscoveryBehaviour` from this config.
|
||||
pub fn finish(self) -> DiscoveryBehaviour {
|
||||
let DiscoveryConfig {
|
||||
let Self {
|
||||
local_peer_id,
|
||||
permanent_addresses,
|
||||
dht_random_walk,
|
||||
@@ -205,8 +205,8 @@ impl DiscoveryConfig {
|
||||
config.set_kbucket_inserts(KademliaBucketInserts::Manual);
|
||||
config.disjoint_query_paths(kademlia_disjoint_query_paths);
|
||||
|
||||
let store = MemoryStore::new(local_peer_id.clone());
|
||||
let mut kad = Kademlia::with_config(local_peer_id.clone(), store, config);
|
||||
let store = MemoryStore::new(local_peer_id);
|
||||
let mut kad = Kademlia::with_config(local_peer_id, store, config);
|
||||
|
||||
for (peer_id, addr) in &permanent_addresses {
|
||||
kad.add_address(peer_id, addr.clone());
|
||||
@@ -324,7 +324,7 @@ impl DiscoveryBehaviour {
|
||||
addr: Multiaddr,
|
||||
) {
|
||||
if !self.allow_non_globals_in_dht && !self.can_add_to_dht(&addr) {
|
||||
log::trace!(target: "sub-libp2p", "Ignoring self-reported non-global address {} from {}.", addr, peer_id);
|
||||
trace!(target: "sub-libp2p", "Ignoring self-reported non-global address {} from {}.", addr, peer_id);
|
||||
return
|
||||
}
|
||||
|
||||
@@ -332,7 +332,7 @@ impl DiscoveryBehaviour {
|
||||
for protocol in supported_protocols {
|
||||
for kademlia in self.kademlias.values_mut() {
|
||||
if protocol.as_ref() == kademlia.protocol_name() {
|
||||
log::trace!(
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"Adding self-reported address {} from {} to Kademlia DHT {}.",
|
||||
addr, peer_id, String::from_utf8_lossy(kademlia.protocol_name()),
|
||||
@@ -344,7 +344,7 @@ impl DiscoveryBehaviour {
|
||||
}
|
||||
|
||||
if !added {
|
||||
log::trace!(
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"Ignoring self-reported address {} from {} as remote node is not part of any \
|
||||
Kademlia DHTs supported by the local node.", addr, peer_id,
|
||||
@@ -593,18 +593,21 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
if let Some(kad) = self.kademlias.get_mut(&pid) {
|
||||
return kad.inject_event(peer_id, connection, event)
|
||||
}
|
||||
log::error!(target: "sub-libp2p",
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"inject_node_event: no kademlia instance registered for protocol {:?}",
|
||||
pid)
|
||||
pid,
|
||||
)
|
||||
}
|
||||
|
||||
fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
|
||||
let new_addr = addr.clone().with(Protocol::P2p(self.local_peer_id.clone().into()));
|
||||
let new_addr = addr.clone().with(Protocol::P2p(self.local_peer_id.into()));
|
||||
|
||||
// NOTE: we might re-discover the same address multiple times
|
||||
// in which case we just want to refrain from logging.
|
||||
if self.known_external_addresses.insert(new_addr.clone()) {
|
||||
info!(target: "sub-libp2p",
|
||||
info!(
|
||||
target: "sub-libp2p",
|
||||
"🔍 Discovered new external address for our node: {}",
|
||||
new_addr,
|
||||
);
|
||||
@@ -680,11 +683,13 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
while let Poll::Ready(_) = next_kad_random_query.poll_unpin(cx) {
|
||||
let actually_started = if self.num_connections < self.discovery_only_if_under_num {
|
||||
let random_peer_id = PeerId::random();
|
||||
debug!(target: "sub-libp2p",
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p <= Starting random Kademlia request for {:?}",
|
||||
random_peer_id);
|
||||
random_peer_id,
|
||||
);
|
||||
for k in self.kademlias.values_mut() {
|
||||
k.get_closest_peers(random_peer_id.clone());
|
||||
k.get_closest_peers(random_peer_id);
|
||||
}
|
||||
true
|
||||
} else {
|
||||
@@ -736,17 +741,23 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
..
|
||||
} => match res {
|
||||
Err(GetClosestPeersError::Timeout { key, peers }) => {
|
||||
debug!(target: "sub-libp2p",
|
||||
"Libp2p => Query for {:?} timed out with {} results",
|
||||
HexDisplay::from(&key), peers.len());
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Query for {:?} timed out with {} results",
|
||||
HexDisplay::from(&key), peers.len(),
|
||||
);
|
||||
},
|
||||
Ok(ok) => {
|
||||
trace!(target: "sub-libp2p",
|
||||
"Libp2p => Query for {:?} yielded {:?} results",
|
||||
HexDisplay::from(&ok.key), ok.peers.len());
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Query for {:?} yielded {:?} results",
|
||||
HexDisplay::from(&ok.key), ok.peers.len(),
|
||||
);
|
||||
if ok.peers.is_empty() && self.num_connections != 0 {
|
||||
debug!(target: "sub-libp2p", "Libp2p => Random Kademlia query has yielded empty \
|
||||
results");
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Random Kademlia query has yielded empty results",
|
||||
);
|
||||
}
|
||||
},
|
||||
},
|
||||
@@ -769,16 +780,22 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
)
|
||||
},
|
||||
Err(e @ libp2p::kad::GetRecordError::NotFound { .. }) => {
|
||||
trace!(target: "sub-libp2p",
|
||||
"Libp2p => Failed to get record: {:?}", e);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Failed to get record: {:?}",
|
||||
e,
|
||||
);
|
||||
DiscoveryOut::ValueNotFound(
|
||||
e.into_key(),
|
||||
stats.duration().unwrap_or_else(Default::default),
|
||||
)
|
||||
},
|
||||
Err(e) => {
|
||||
debug!(target: "sub-libp2p",
|
||||
"Libp2p => Failed to get record: {:?}", e);
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Failed to get record: {:?}",
|
||||
e,
|
||||
);
|
||||
DiscoveryOut::ValueNotFound(
|
||||
e.into_key(),
|
||||
stats.duration().unwrap_or_else(Default::default),
|
||||
@@ -798,8 +815,11 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
stats.duration().unwrap_or_else(Default::default),
|
||||
),
|
||||
Err(e) => {
|
||||
debug!(target: "sub-libp2p",
|
||||
"Libp2p => Failed to put record: {:?}", e);
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Failed to put record: {:?}",
|
||||
e,
|
||||
);
|
||||
DiscoveryOut::ValuePutFailed(
|
||||
e.into_key(),
|
||||
stats.duration().unwrap_or_else(Default::default),
|
||||
@@ -812,12 +832,16 @@ impl NetworkBehaviour for DiscoveryBehaviour {
|
||||
result: QueryResult::RepublishRecord(res),
|
||||
..
|
||||
} => match res {
|
||||
Ok(ok) => debug!(target: "sub-libp2p",
|
||||
"Libp2p => Record republished: {:?}",
|
||||
ok.key),
|
||||
Err(e) => debug!(target: "sub-libp2p",
|
||||
"Libp2p => Republishing of record {:?} failed with: {:?}",
|
||||
e.key(), e),
|
||||
Ok(ok) => debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Record republished: {:?}",
|
||||
ok.key,
|
||||
),
|
||||
Err(e) => debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Republishing of record {:?} failed with: {:?}",
|
||||
e.key(), e,
|
||||
),
|
||||
},
|
||||
// We never start any other type of query.
|
||||
e => {
|
||||
@@ -907,9 +931,9 @@ enum MdnsWrapper {
|
||||
impl MdnsWrapper {
|
||||
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
|
||||
match self {
|
||||
MdnsWrapper::Instantiating(_) => Vec::new(),
|
||||
MdnsWrapper::Ready(mdns) => mdns.addresses_of_peer(peer_id),
|
||||
MdnsWrapper::Disabled => Vec::new(),
|
||||
Self::Instantiating(_) => Vec::new(),
|
||||
Self::Ready(mdns) => mdns.addresses_of_peer(peer_id),
|
||||
Self::Disabled => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -920,16 +944,16 @@ impl MdnsWrapper {
|
||||
) -> Poll<NetworkBehaviourAction<void::Void, MdnsEvent>> {
|
||||
loop {
|
||||
match self {
|
||||
MdnsWrapper::Instantiating(fut) =>
|
||||
Self::Instantiating(fut) =>
|
||||
*self = match futures::ready!(fut.as_mut().poll(cx)) {
|
||||
Ok(mdns) => MdnsWrapper::Ready(mdns),
|
||||
Ok(mdns) => Self::Ready(mdns),
|
||||
Err(err) => {
|
||||
warn!(target: "sub-libp2p", "Failed to initialize mDNS: {:?}", err);
|
||||
MdnsWrapper::Disabled
|
||||
Self::Disabled
|
||||
},
|
||||
},
|
||||
MdnsWrapper::Ready(mdns) => return mdns.poll(cx, params),
|
||||
MdnsWrapper::Disabled => return Poll::Pending,
|
||||
Self::Ready(mdns) => return mdns.poll(cx, params),
|
||||
Self::Disabled => return Poll::Pending,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1100,7 +1124,7 @@ mod tests {
|
||||
for kademlia in discovery.kademlias.values_mut() {
|
||||
assert!(
|
||||
kademlia
|
||||
.kbucket(remote_peer_id.clone())
|
||||
.kbucket(remote_peer_id)
|
||||
.expect("Remote peer id not to be equal to local peer id.")
|
||||
.is_empty(),
|
||||
"Expect peer with unsupported protocol not to be added."
|
||||
@@ -1118,7 +1142,7 @@ mod tests {
|
||||
assert_eq!(
|
||||
1,
|
||||
kademlia
|
||||
.kbucket(remote_peer_id.clone())
|
||||
.kbucket(remote_peer_id)
|
||||
.expect("Remote peer id not to be equal to local peer id.")
|
||||
.num_entries(),
|
||||
"Expect peer with supported protocol to be added."
|
||||
@@ -1159,7 +1183,7 @@ mod tests {
|
||||
.kademlias
|
||||
.get_mut(&protocol_a)
|
||||
.expect("Kademlia instance to exist.")
|
||||
.kbucket(remote_peer_id.clone())
|
||||
.kbucket(remote_peer_id)
|
||||
.expect("Remote peer id not to be equal to local peer id.")
|
||||
.num_entries(),
|
||||
"Expected remote peer to be added to `protocol_a` Kademlia instance.",
|
||||
@@ -1170,7 +1194,7 @@ mod tests {
|
||||
.kademlias
|
||||
.get_mut(&protocol_b)
|
||||
.expect("Kademlia instance to exist.")
|
||||
.kbucket(remote_peer_id.clone())
|
||||
.kbucket(remote_peer_id)
|
||||
.expect("Remote peer id not to be equal to local peer id.")
|
||||
.is_empty(),
|
||||
"Expected remote peer not to be added to `protocol_b` Kademlia instance.",
|
||||
|
||||
@@ -79,12 +79,12 @@ impl fmt::Debug for Error {
|
||||
impl std::error::Error for Error {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
Error::Io(ref err) => Some(err),
|
||||
Error::Client(ref err) => Some(err),
|
||||
Error::DuplicateBootnode { .. } => None,
|
||||
Error::Prometheus(ref err) => Some(err),
|
||||
Error::AddressesForAnotherTransport { .. } => None,
|
||||
Error::DuplicateRequestResponseProtocol { .. } => None,
|
||||
Self::Io(ref err) => Some(err),
|
||||
Self::Client(ref err) => Some(err),
|
||||
Self::Prometheus(ref err) => Some(err),
|
||||
Self::DuplicateBootnode { .. } |
|
||||
Self::AddressesForAnotherTransport { .. } |
|
||||
Self::DuplicateRequestResponseProtocol { .. } => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -108,7 +108,7 @@ impl<M> QueuedSender<M> {
|
||||
messages_encode
|
||||
);
|
||||
|
||||
let sender = QueuedSender {
|
||||
let sender = Self {
|
||||
shared_message_queue,
|
||||
notify_background_future,
|
||||
queue_size_limit,
|
||||
|
||||
@@ -158,12 +158,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
peer: &PeerId,
|
||||
request: &schema::v1::light::RemoteCallRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
log::trace!(
|
||||
"Remote call request from {} ({} at {:?}).",
|
||||
peer,
|
||||
request.method,
|
||||
request.block,
|
||||
);
|
||||
trace!("Remote call request from {} ({} at {:?}).", peer, request.method, request.block,);
|
||||
|
||||
let block = Decode::decode(&mut request.block.as_ref())?;
|
||||
|
||||
@@ -174,7 +169,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
{
|
||||
Ok((_, proof)) => proof,
|
||||
Err(e) => {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"remote call request from {} ({} at {:?}) failed with: {}",
|
||||
peer,
|
||||
request.method,
|
||||
@@ -199,11 +194,11 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
request: &schema::v1::light::RemoteReadRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
if request.keys.is_empty() {
|
||||
log::debug!("Invalid remote read request sent by {}.", peer);
|
||||
debug!("Invalid remote read request sent by {}.", peer);
|
||||
return Err(HandleRequestError::BadRequest("Remote read request without keys."))
|
||||
}
|
||||
|
||||
log::trace!(
|
||||
trace!(
|
||||
"Remote read request from {} ({} at {:?}).",
|
||||
peer,
|
||||
fmt_keys(request.keys.first(), request.keys.last()),
|
||||
@@ -218,7 +213,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
{
|
||||
Ok(proof) => proof,
|
||||
Err(error) => {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"remote read request from {} ({} at {:?}) failed with: {}",
|
||||
peer,
|
||||
fmt_keys(request.keys.first(), request.keys.last()),
|
||||
@@ -243,11 +238,11 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
request: &schema::v1::light::RemoteReadChildRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
if request.keys.is_empty() {
|
||||
log::debug!("Invalid remote child read request sent by {}.", peer);
|
||||
debug!("Invalid remote child read request sent by {}.", peer);
|
||||
return Err(HandleRequestError::BadRequest("Remove read child request without keys."))
|
||||
}
|
||||
|
||||
log::trace!(
|
||||
trace!(
|
||||
"Remote read child request from {} ({} {} at {:?}).",
|
||||
peer,
|
||||
HexDisplay::from(&request.storage_key),
|
||||
@@ -271,7 +266,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
}) {
|
||||
Ok(proof) => proof,
|
||||
Err(error) => {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"remote read child request from {} ({} {} at {:?}) failed with: {}",
|
||||
peer,
|
||||
HexDisplay::from(&request.storage_key),
|
||||
@@ -296,13 +291,13 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
peer: &PeerId,
|
||||
request: &schema::v1::light::RemoteHeaderRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
log::trace!("Remote header proof request from {} ({:?}).", peer, request.block);
|
||||
trace!("Remote header proof request from {} ({:?}).", peer, request.block);
|
||||
|
||||
let block = Decode::decode(&mut request.block.as_ref())?;
|
||||
let (header, proof) = match self.client.header_proof(&BlockId::Number(block)) {
|
||||
Ok((header, proof)) => (header.encode(), proof),
|
||||
Err(error) => {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"Remote header proof request from {} ({:?}) failed with: {}.",
|
||||
peer,
|
||||
request.block,
|
||||
@@ -325,7 +320,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
peer: &PeerId,
|
||||
request: &schema::v1::light::RemoteChangesRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"Remote changes proof request from {} for key {} ({:?}..{:?}).",
|
||||
peer,
|
||||
if !request.storage_key.is_empty() {
|
||||
@@ -356,7 +351,7 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
match self.client.key_changes_proof(first, last, min, max, storage_key, &key) {
|
||||
Ok(proof) => proof,
|
||||
Err(error) => {
|
||||
log::trace!(
|
||||
trace!(
|
||||
"Remote changes proof request from {} for key {} ({:?}..{:?}) failed with: {}.",
|
||||
peer,
|
||||
format!("{} : {}", HexDisplay::from(&request.storage_key), HexDisplay::from(&key.0)),
|
||||
|
||||
@@ -72,7 +72,7 @@ struct Config {
|
||||
impl Config {
|
||||
/// Create a new [`LightClientRequestSender`] configuration.
|
||||
pub fn new(id: &ProtocolId) -> Self {
|
||||
Config {
|
||||
Self {
|
||||
max_pending_requests: 128,
|
||||
light_protocol: super::generate_protocol_name(id),
|
||||
block_protocol: crate::block_request_handler::generate_protocol_name(id),
|
||||
@@ -112,7 +112,7 @@ struct PendingRequest<B: Block> {
|
||||
|
||||
impl<B: Block> PendingRequest<B> {
|
||||
fn new(req: Request<B>) -> Self {
|
||||
PendingRequest {
|
||||
Self {
|
||||
// Number of retries + one for the initial attempt.
|
||||
attempts_left: req.retries() + 1,
|
||||
request: req,
|
||||
@@ -153,7 +153,7 @@ where
|
||||
checker: Arc<dyn light::FetchChecker<B>>,
|
||||
peerset: sc_peerset::PeersetHandle,
|
||||
) -> Self {
|
||||
LightClientRequestSender {
|
||||
Self {
|
||||
config: Config::new(id),
|
||||
checker,
|
||||
peers: Default::default(),
|
||||
|
||||
@@ -93,9 +93,9 @@ pub enum PeerEndpoint {
|
||||
impl From<ConnectedPoint> for PeerEndpoint {
|
||||
fn from(endpoint: ConnectedPoint) -> Self {
|
||||
match endpoint {
|
||||
ConnectedPoint::Dialer { address } => PeerEndpoint::Dialing(address),
|
||||
ConnectedPoint::Dialer { address } => Self::Dialing(address),
|
||||
ConnectedPoint::Listener { local_addr, send_back_addr } =>
|
||||
PeerEndpoint::Listening { local_addr, send_back_addr },
|
||||
Self::Listening { local_addr, send_back_addr },
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,7 +135,7 @@ where
|
||||
let (requests_send, requests_queue) = tracing_unbounded("mpsc_ondemand");
|
||||
let requests_queue = Mutex::new(Some(requests_queue));
|
||||
|
||||
OnDemand { checker, requests_queue, requests_send }
|
||||
Self { checker, requests_queue, requests_send }
|
||||
}
|
||||
|
||||
/// Get checker reference.
|
||||
|
||||
@@ -78,7 +78,7 @@ impl NodeInfo {
|
||||
fn new(endpoint: ConnectedPoint) -> Self {
|
||||
let mut endpoints = SmallVec::new();
|
||||
endpoints.push(endpoint);
|
||||
NodeInfo { info_expire: None, endpoints, client_version: None, latest_ping: None }
|
||||
Self { info_expire: None, endpoints, client_version: None, latest_ping: None }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,7 +91,7 @@ impl PeerInfoBehaviour {
|
||||
Identify::new(cfg)
|
||||
};
|
||||
|
||||
PeerInfoBehaviour {
|
||||
Self {
|
||||
ping: Ping::new(PingConfig::new()),
|
||||
identify,
|
||||
nodes_info: FnvHashMap::default(),
|
||||
@@ -199,7 +199,7 @@ impl NetworkBehaviour for PeerInfoBehaviour {
|
||||
) {
|
||||
self.ping.inject_connection_established(peer_id, conn, endpoint);
|
||||
self.identify.inject_connection_established(peer_id, conn, endpoint);
|
||||
match self.nodes_info.entry(peer_id.clone()) {
|
||||
match self.nodes_info.entry(*peer_id) {
|
||||
Entry::Vacant(e) => {
|
||||
e.insert(NodeInfo::new(endpoint.clone()));
|
||||
},
|
||||
|
||||
@@ -41,7 +41,7 @@ use libp2p::{
|
||||
},
|
||||
Multiaddr, PeerId,
|
||||
};
|
||||
use log::{debug, error, log, trace, warn, Level};
|
||||
use log::{debug, error, info, log, trace, warn, Level};
|
||||
use message::{
|
||||
generic::{Message as GenericMessage, Roles},
|
||||
BlockAnnounce, Message,
|
||||
@@ -130,7 +130,7 @@ struct Metrics {
|
||||
|
||||
impl Metrics {
|
||||
fn register(r: &Registry) -> Result<Self, PrometheusError> {
|
||||
Ok(Metrics {
|
||||
Ok(Self {
|
||||
peers: {
|
||||
let g = Gauge::new("sync_peers", "Number of peers we sync with")?;
|
||||
register(g, r)?
|
||||
@@ -249,11 +249,7 @@ impl ProtocolConfig {
|
||||
|
||||
impl Default for ProtocolConfig {
|
||||
fn default() -> ProtocolConfig {
|
||||
ProtocolConfig {
|
||||
roles: Roles::FULL,
|
||||
max_parallel_downloads: 5,
|
||||
sync_mode: config::SyncMode::Full,
|
||||
}
|
||||
Self { roles: Roles::FULL, max_parallel_downloads: 5, sync_mode: config::SyncMode::Full }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -277,12 +273,7 @@ impl<B: BlockT> BlockAnnouncesHandshake<B> {
|
||||
best_hash: B::Hash,
|
||||
genesis_hash: B::Hash,
|
||||
) -> Self {
|
||||
BlockAnnouncesHandshake {
|
||||
genesis_hash,
|
||||
roles: protocol_config.roles,
|
||||
best_number,
|
||||
best_hash,
|
||||
}
|
||||
Self { genesis_hash, roles: protocol_config.roles, best_number, best_hash }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -311,7 +302,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
let boot_node_ids = {
|
||||
let mut list = HashSet::new();
|
||||
for node in &network_config.boot_nodes {
|
||||
list.insert(node.peer_id.clone());
|
||||
list.insert(node.peer_id);
|
||||
}
|
||||
list.shrink_to_fit();
|
||||
list
|
||||
@@ -320,14 +311,14 @@ impl<B: BlockT> Protocol<B> {
|
||||
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());
|
||||
imp_p.insert(reserved.peer_id);
|
||||
}
|
||||
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.insert(reserved.peer_id);
|
||||
}
|
||||
imp_p.shrink_to_fit();
|
||||
imp_p
|
||||
@@ -341,14 +332,14 @@ impl<B: BlockT> Protocol<B> {
|
||||
|
||||
let mut default_sets_reserved = HashSet::new();
|
||||
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()));
|
||||
default_sets_reserved.insert(reserved.peer_id);
|
||||
known_addresses.push((reserved.peer_id, 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()));
|
||||
bootnodes.push(bootnode.peer_id);
|
||||
known_addresses.push((bootnode.peer_id, bootnode.multiaddr.clone()));
|
||||
}
|
||||
|
||||
// Set number 0 is used for block announces.
|
||||
@@ -364,8 +355,8 @@ impl<B: BlockT> Protocol<B> {
|
||||
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()));
|
||||
reserved_nodes.insert(reserved.peer_id);
|
||||
known_addresses.push((reserved.peer_id, reserved.multiaddr.clone()));
|
||||
}
|
||||
|
||||
let reserved_only =
|
||||
@@ -427,7 +418,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
network_config.default_peers_set.out_peers as usize,
|
||||
);
|
||||
|
||||
let protocol = Protocol {
|
||||
let protocol = Self {
|
||||
tick_timeout: Box::pin(interval(TICK_TIMEOUT)),
|
||||
pending_messages: VecDeque::new(),
|
||||
config,
|
||||
@@ -481,7 +472,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
sc_peerset::SetId::from(position + NUM_HARDCODED_PEERSETS),
|
||||
);
|
||||
} else {
|
||||
log::warn!(target: "sub-libp2p", "disconnect_peer() with invalid protocol name")
|
||||
warn!(target: "sub-libp2p", "disconnect_peer() with invalid protocol name")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -769,7 +760,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
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);
|
||||
error!(target: "sync", "Called on_sync_peer_connected with already connected peer {}", who);
|
||||
debug_assert!(false);
|
||||
return Err(())
|
||||
}
|
||||
@@ -781,7 +772,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
"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.peerset_handle.report_peer(who, rep::GENESIS_MISMATCH);
|
||||
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
|
||||
|
||||
if self.boot_node_ids.contains(&who) {
|
||||
@@ -801,7 +792,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
// we're not interested in light peers
|
||||
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.peerset_handle.report_peer(who, rep::BAD_ROLE);
|
||||
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
|
||||
return Err(())
|
||||
}
|
||||
@@ -814,7 +805,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
.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.peerset_handle.report_peer(who, rep::PEER_BEHIND_US_LIGHT);
|
||||
self.behaviour.disconnect_peer(&who, HARDCODED_PEERSETS_SYNC);
|
||||
return Err(())
|
||||
}
|
||||
@@ -833,7 +824,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
};
|
||||
|
||||
let req = if peer.info.roles.is_full() {
|
||||
match self.sync.new_peer(who.clone(), peer.info.best_hash, peer.info.best_number) {
|
||||
match self.sync.new_peer(who, peer.info.best_hash, peer.info.best_number) {
|
||||
Ok(req) => req,
|
||||
Err(sync::BadPeer(id, repu)) => {
|
||||
self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC);
|
||||
@@ -847,12 +838,12 @@ impl<B: BlockT> Protocol<B> {
|
||||
|
||||
debug!(target: "sync", "Connected {}", who);
|
||||
|
||||
self.peers.insert(who.clone(), peer);
|
||||
self.peers.insert(who, peer);
|
||||
self.pending_messages
|
||||
.push_back(CustomMessageOutcome::PeerNewBest(who.clone(), status.best_number));
|
||||
.push_back(CustomMessageOutcome::PeerNewBest(who, status.best_number));
|
||||
|
||||
if let Some(req) = req {
|
||||
let event = self.prepare_block_request(who.clone(), req);
|
||||
let event = self.prepare_block_request(who, req);
|
||||
self.pending_messages.push_back(event);
|
||||
}
|
||||
|
||||
@@ -1101,7 +1092,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
) {
|
||||
self.sync.on_justification_import(hash, number, success);
|
||||
if !success {
|
||||
log::info!("💔 Invalid justification provided by {} for #{}", who, hash);
|
||||
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"));
|
||||
@@ -1141,7 +1132,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
peer,
|
||||
);
|
||||
} else {
|
||||
log::error!(
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"remove_set_reserved_peer with unknown protocol: {}",
|
||||
protocol
|
||||
@@ -1155,7 +1146,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
self.peerset_handle
|
||||
.add_reserved_peer(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
|
||||
} else {
|
||||
log::error!(
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"add_set_reserved_peer with unknown protocol: {}",
|
||||
protocol
|
||||
@@ -1178,7 +1169,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
self.peerset_handle
|
||||
.add_to_peers_set(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
|
||||
} else {
|
||||
log::error!(
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"add_to_peers_set with unknown protocol: {}",
|
||||
protocol
|
||||
@@ -1194,7 +1185,7 @@ impl<B: BlockT> Protocol<B> {
|
||||
peer,
|
||||
);
|
||||
} else {
|
||||
log::error!(
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"remove_from_peers_set with unknown protocol: {}",
|
||||
protocol
|
||||
@@ -1426,8 +1417,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
id,
|
||||
e
|
||||
);
|
||||
self.peerset_handle
|
||||
.report_peer(id.clone(), rep::BAD_MESSAGE);
|
||||
self.peerset_handle.report_peer(*id, rep::BAD_MESSAGE);
|
||||
self.behaviour
|
||||
.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
|
||||
continue
|
||||
@@ -1447,18 +1437,17 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
id,
|
||||
e
|
||||
);
|
||||
self.peerset_handle
|
||||
.report_peer(id.clone(), rep::BAD_MESSAGE);
|
||||
self.peerset_handle.report_peer(*id, rep::BAD_MESSAGE);
|
||||
self.behaviour
|
||||
.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
|
||||
continue
|
||||
},
|
||||
};
|
||||
|
||||
finished_state_requests.push((id.clone(), protobuf_response));
|
||||
finished_state_requests.push((*id, protobuf_response));
|
||||
},
|
||||
PeerRequest::WarpProof => {
|
||||
finished_warp_sync_requests.push((id.clone(), resp));
|
||||
finished_warp_sync_requests.push((*id, resp));
|
||||
},
|
||||
}
|
||||
},
|
||||
@@ -1468,18 +1457,18 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
|
||||
match e {
|
||||
RequestFailure::Network(OutboundFailure::Timeout) => {
|
||||
self.peerset_handle.report_peer(id.clone(), rep::TIMEOUT);
|
||||
self.peerset_handle.report_peer(*id, 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.peerset_handle.report_peer(*id, 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.peerset_handle.report_peer(*id, rep::REFUSED);
|
||||
self.behaviour.disconnect_peer(id, HARDCODED_PEERSETS_SYNC);
|
||||
},
|
||||
RequestFailure::Network(OutboundFailure::ConnectionClosed) |
|
||||
@@ -1603,7 +1592,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
genesis_hash: handshake.genesis_hash,
|
||||
};
|
||||
|
||||
if self.on_sync_peer_connected(peer_id.clone(), handshake).is_ok() {
|
||||
if self.on_sync_peer_connected(peer_id, handshake).is_ok() {
|
||||
CustomMessageOutcome::SyncConnected(peer_id)
|
||||
} else {
|
||||
CustomMessageOutcome::None
|
||||
@@ -1624,10 +1613,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
&mut &received_handshake[..],
|
||||
) {
|
||||
Ok(handshake) => {
|
||||
if self
|
||||
.on_sync_peer_connected(peer_id.clone(), handshake)
|
||||
.is_ok()
|
||||
{
|
||||
if self.on_sync_peer_connected(peer_id, handshake).is_ok() {
|
||||
CustomMessageOutcome::SyncConnected(peer_id)
|
||||
} else {
|
||||
CustomMessageOutcome::None
|
||||
@@ -1679,7 +1665,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
},
|
||||
(Err(err), _) => {
|
||||
debug!(target: "sync", "Failed to parse remote handshake: {}", err);
|
||||
self.bad_handshake_substreams.insert((peer_id.clone(), set_id));
|
||||
self.bad_handshake_substreams.insert((peer_id, set_id));
|
||||
self.behaviour.disconnect_peer(&peer_id, set_id);
|
||||
self.peerset_handle.report_peer(peer_id, rep::BAD_MESSAGE);
|
||||
CustomMessageOutcome::None
|
||||
@@ -1690,7 +1676,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
NotificationsOut::CustomProtocolReplaced { peer_id, notifications_sink, set_id } =>
|
||||
if set_id == HARDCODED_PEERSETS_SYNC {
|
||||
CustomMessageOutcome::None
|
||||
} else if self.bad_handshake_substreams.contains(&(peer_id.clone(), set_id)) {
|
||||
} else if self.bad_handshake_substreams.contains(&(peer_id, set_id)) {
|
||||
CustomMessageOutcome::None
|
||||
} else {
|
||||
CustomMessageOutcome::NotificationStreamReplaced {
|
||||
@@ -1704,7 +1690,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
NotificationsOut::CustomProtocolClosed { peer_id, set_id } => {
|
||||
// Set number 0 is hardcoded the default set of peers we sync from.
|
||||
if set_id == HARDCODED_PEERSETS_SYNC {
|
||||
if self.on_sync_peer_disconnected(peer_id.clone()).is_ok() {
|
||||
if self.on_sync_peer_disconnected(peer_id).is_ok() {
|
||||
CustomMessageOutcome::SyncDisconnected(peer_id)
|
||||
} else {
|
||||
log::trace!(
|
||||
@@ -1714,7 +1700,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
);
|
||||
CustomMessageOutcome::None
|
||||
}
|
||||
} else if self.bad_handshake_substreams.remove(&(peer_id.clone(), set_id)) {
|
||||
} else if self.bad_handshake_substreams.remove(&(peer_id, set_id)) {
|
||||
// The substream that has just been closed had been opened with a bad
|
||||
// handshake. The outer layers have never received an opening event about this
|
||||
// substream, and consequently shouldn't receive a closing event either.
|
||||
@@ -1753,7 +1739,7 @@ impl<B: BlockT> NetworkBehaviour for Protocol<B> {
|
||||
);
|
||||
CustomMessageOutcome::None
|
||||
},
|
||||
_ if self.bad_handshake_substreams.contains(&(peer_id.clone(), set_id)) =>
|
||||
_ if self.bad_handshake_substreams.contains(&(peer_id, set_id)) =>
|
||||
CustomMessageOutcome::None,
|
||||
_ => {
|
||||
let protocol_name = self.notification_protocols
|
||||
|
||||
@@ -117,6 +117,6 @@ pub enum ObservedRole {
|
||||
impl ObservedRole {
|
||||
/// Returns `true` for `ObservedRole::Light`.
|
||||
pub fn is_light(&self) -> bool {
|
||||
matches!(self, ObservedRole::Light)
|
||||
matches!(self, Self::Light)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -85,7 +85,7 @@ impl BlockAttributes {
|
||||
|
||||
/// Decodes attributes, encoded with the `encode_to_be_u32()` call.
|
||||
pub fn from_be_u32(encoded: u32) -> Result<Self, Error> {
|
||||
BlockAttributes::from_bits(encoded.to_be_bytes()[0])
|
||||
Self::from_bits(encoded.to_be_bytes()[0])
|
||||
.ok_or_else(|| Error::from("Invalid BlockAttributes"))
|
||||
}
|
||||
}
|
||||
@@ -187,12 +187,12 @@ pub mod generic {
|
||||
impl Roles {
|
||||
/// Does this role represents a client that holds full chain data locally?
|
||||
pub fn is_full(&self) -> bool {
|
||||
self.intersects(Roles::FULL | Roles::AUTHORITY)
|
||||
self.intersects(Self::FULL | Self::AUTHORITY)
|
||||
}
|
||||
|
||||
/// Does this role represents a client that does not participates in the consensus?
|
||||
pub fn is_authority(&self) -> bool {
|
||||
*self == Roles::AUTHORITY
|
||||
*self == Self::AUTHORITY
|
||||
}
|
||||
|
||||
/// Does this role represents a client that does not hold full chain data locally?
|
||||
@@ -204,9 +204,9 @@ pub mod generic {
|
||||
impl<'a> From<&'a crate::config::Role> for Roles {
|
||||
fn from(roles: &'a crate::config::Role) -> Self {
|
||||
match roles {
|
||||
crate::config::Role::Full => Roles::FULL,
|
||||
crate::config::Role::Light => Roles::LIGHT,
|
||||
crate::config::Role::Authority { .. } => Roles::AUTHORITY,
|
||||
crate::config::Role::Full => Self::FULL,
|
||||
crate::config::Role::Light => Self::LIGHT,
|
||||
crate::config::Role::Authority { .. } => Self::AUTHORITY,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -368,7 +368,7 @@ pub mod generic {
|
||||
genesis_hash,
|
||||
} = compact;
|
||||
|
||||
Ok(Status {
|
||||
Ok(Self {
|
||||
version,
|
||||
min_supported_version,
|
||||
roles,
|
||||
@@ -438,7 +438,7 @@ pub mod generic {
|
||||
let header = H::decode(input)?;
|
||||
let state = BlockState::decode(input).ok();
|
||||
let data = Vec::decode(input).ok();
|
||||
Ok(BlockAnnounce { header, state, data })
|
||||
Ok(Self { header, state, data })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ use libp2p::{
|
||||
use log::{error, trace, warn};
|
||||
use parking_lot::RwLock;
|
||||
use rand::distributions::{Distribution as _, Uniform};
|
||||
use sc_peerset::DropReason;
|
||||
use smallvec::SmallVec;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
@@ -242,35 +243,22 @@ impl PeerState {
|
||||
/// that is open for custom protocol traffic.
|
||||
fn get_open(&self) -> Option<&NotificationsSink> {
|
||||
match self {
|
||||
PeerState::Enabled { connections, .. } => connections
|
||||
.iter()
|
||||
.filter_map(|(_, s)| match s {
|
||||
ConnectionState::Open(s) => Some(s),
|
||||
_ => None,
|
||||
})
|
||||
.next(),
|
||||
PeerState::Poisoned => None,
|
||||
PeerState::Backoff { .. } => None,
|
||||
PeerState::PendingRequest { .. } => None,
|
||||
PeerState::Requested => None,
|
||||
PeerState::Disabled { .. } => None,
|
||||
PeerState::DisabledPendingEnable { .. } => None,
|
||||
PeerState::Incoming { .. } => None,
|
||||
Self::Enabled { connections, .. } => connections.iter().find_map(|(_, s)| match s {
|
||||
ConnectionState::Open(s) => Some(s),
|
||||
_ => None,
|
||||
}),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// True if that node has been requested by the PSM.
|
||||
fn is_requested(&self) -> bool {
|
||||
match self {
|
||||
PeerState::Poisoned => false,
|
||||
PeerState::Backoff { .. } => false,
|
||||
PeerState::PendingRequest { .. } => true,
|
||||
PeerState::Requested => true,
|
||||
PeerState::Disabled { .. } => false,
|
||||
PeerState::DisabledPendingEnable { .. } => true,
|
||||
PeerState::Enabled { .. } => true,
|
||||
PeerState::Incoming { .. } => false,
|
||||
}
|
||||
matches!(
|
||||
self,
|
||||
Self::PendingRequest { .. } |
|
||||
Self::Requested | Self::DisabledPendingEnable { .. } |
|
||||
Self::Enabled { .. }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -391,7 +379,7 @@ impl Notifications {
|
||||
|
||||
assert!(!notif_protocols.is_empty());
|
||||
|
||||
Notifications {
|
||||
Self {
|
||||
notif_protocols,
|
||||
peerset,
|
||||
peers: FnvHashMap::default(),
|
||||
@@ -446,8 +434,7 @@ impl Notifications {
|
||||
set_id: sc_peerset::SetId,
|
||||
ban: Option<Duration>,
|
||||
) {
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((peer_id.clone(), set_id))
|
||||
{
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((*peer_id, set_id)) {
|
||||
entry
|
||||
} else {
|
||||
return
|
||||
@@ -463,7 +450,7 @@ impl Notifications {
|
||||
// DisabledPendingEnable => Disabled.
|
||||
PeerState::DisabledPendingEnable { connections, timer_deadline, timer: _ } => {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
let backoff_until = Some(if let Some(ban) = ban {
|
||||
cmp::max(timer_deadline, Instant::now() + ban)
|
||||
} else {
|
||||
@@ -477,12 +464,12 @@ impl Notifications {
|
||||
// If relevant, the external API is instantly notified.
|
||||
PeerState::Enabled { mut connections } => {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
|
||||
if connections.iter().any(|(_, s)| matches!(s, ConnectionState::Open(_))) {
|
||||
trace!(target: "sub-libp2p", "External API <= Closed({}, {:?})", peer_id, set_id);
|
||||
let event =
|
||||
NotificationsOut::CustomProtocolClosed { peer_id: peer_id.clone(), set_id };
|
||||
NotificationsOut::CustomProtocolClosed { peer_id: *peer_id, set_id };
|
||||
self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
|
||||
}
|
||||
|
||||
@@ -491,7 +478,7 @@ impl Notifications {
|
||||
{
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -503,7 +490,7 @@ impl Notifications {
|
||||
{
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -531,8 +518,10 @@ impl Notifications {
|
||||
{
|
||||
inc
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "State mismatch in libp2p: no entry in \
|
||||
incoming for incoming peer");
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"State mismatch in libp2p: no entry in incoming for incoming peer"
|
||||
);
|
||||
return
|
||||
};
|
||||
|
||||
@@ -544,7 +533,7 @@ impl Notifications {
|
||||
{
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -605,12 +594,13 @@ impl Notifications {
|
||||
set_id: sc_peerset::SetId,
|
||||
message: impl Into<Vec<u8>>,
|
||||
) {
|
||||
let notifs_sink = match self.peers.get(&(target.clone(), set_id)).and_then(|p| p.get_open())
|
||||
{
|
||||
let notifs_sink = match self.peers.get(&(*target, set_id)).and_then(|p| p.get_open()) {
|
||||
None => {
|
||||
trace!(target: "sub-libp2p",
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"Tried to sent notification to {:?} without an open channel.",
|
||||
target);
|
||||
target,
|
||||
);
|
||||
return
|
||||
},
|
||||
Some(sink) => sink,
|
||||
@@ -638,12 +628,16 @@ impl Notifications {
|
||||
/// Function that is called when the peerset wants us to connect to a peer.
|
||||
fn peerset_report_connect(&mut self, peer_id: PeerId, set_id: sc_peerset::SetId) {
|
||||
// If `PeerId` is unknown to us, insert an entry, start dialing, and return early.
|
||||
let mut occ_entry = match self.peers.entry((peer_id.clone(), set_id)) {
|
||||
let mut occ_entry = match self.peers.entry((peer_id, set_id)) {
|
||||
Entry::Occupied(entry) => entry,
|
||||
Entry::Vacant(entry) => {
|
||||
// If there's no entry in `self.peers`, start dialing.
|
||||
trace!(target: "sub-libp2p", "PSM => Connect({}, {:?}): Starting to connect",
|
||||
entry.key().0, set_id);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"PSM => Connect({}, {:?}): Starting to connect",
|
||||
entry.key().0,
|
||||
set_id,
|
||||
);
|
||||
trace!(target: "sub-libp2p", "Libp2p <= Dial {}", entry.key().0);
|
||||
// The `DialPeerCondition` ensures that dial attempts are de-duplicated
|
||||
self.events.push_back(NetworkBehaviourAction::DialPeer {
|
||||
@@ -661,16 +655,25 @@ impl Notifications {
|
||||
// Backoff (not expired) => PendingRequest
|
||||
PeerState::Backoff { ref timer, ref timer_deadline } if *timer_deadline > now => {
|
||||
let peer_id = occ_entry.key().0.clone();
|
||||
trace!(target: "sub-libp2p", "PSM => Connect({}, {:?}): Will start to connect at \
|
||||
until {:?}", peer_id, set_id, timer_deadline);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"PSM => Connect({}, {:?}): Will start to connect at until {:?}",
|
||||
peer_id,
|
||||
set_id,
|
||||
timer_deadline,
|
||||
);
|
||||
*occ_entry.into_mut() =
|
||||
PeerState::PendingRequest { timer: *timer, timer_deadline: *timer_deadline };
|
||||
},
|
||||
|
||||
// Backoff (expired) => Requested
|
||||
PeerState::Backoff { .. } => {
|
||||
trace!(target: "sub-libp2p", "PSM => Connect({}, {:?}): Starting to connect",
|
||||
occ_entry.key().0, set_id);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"PSM => Connect({}, {:?}): Starting to connect",
|
||||
occ_entry.key().0,
|
||||
set_id,
|
||||
);
|
||||
trace!(target: "sub-libp2p", "Libp2p <= Dial {:?}", occ_entry.key());
|
||||
// The `DialPeerCondition` ensures that dial attempts are de-duplicated
|
||||
self.events.push_back(NetworkBehaviourAction::DialPeer {
|
||||
@@ -685,8 +688,13 @@ impl Notifications {
|
||||
if *backoff > now =>
|
||||
{
|
||||
let peer_id = occ_entry.key().0.clone();
|
||||
trace!(target: "sub-libp2p", "PSM => Connect({}, {:?}): But peer is backed-off until {:?}",
|
||||
peer_id, set_id, backoff);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"PSM => Connect({}, {:?}): But peer is backed-off until {:?}",
|
||||
peer_id,
|
||||
set_id,
|
||||
backoff,
|
||||
);
|
||||
|
||||
let delay_id = self.next_delay_id;
|
||||
self.next_delay_id.0 += 1;
|
||||
@@ -720,7 +728,7 @@ impl Notifications {
|
||||
occ_entry.key().0, set_id);
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})", peer_id, *connec_id, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -778,8 +786,10 @@ impl Notifications {
|
||||
{
|
||||
inc.alive = false;
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "State mismatch in libp2p: no entry in \
|
||||
incoming for incoming peer")
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"State mismatch in libp2p: no entry in incoming for incoming peer",
|
||||
)
|
||||
}
|
||||
|
||||
debug_assert!(connections
|
||||
@@ -953,23 +963,19 @@ impl Notifications {
|
||||
if !incoming.alive {
|
||||
trace!(target: "sub-libp2p", "PSM => Accept({:?}, {}, {:?}): Obsolete incoming",
|
||||
index, incoming.peer_id, incoming.set_id);
|
||||
match self.peers.get_mut(&(incoming.peer_id.clone(), incoming.set_id)) {
|
||||
match self.peers.get_mut(&(incoming.peer_id, incoming.set_id)) {
|
||||
Some(PeerState::DisabledPendingEnable { .. }) | Some(PeerState::Enabled { .. }) => {
|
||||
},
|
||||
_ => {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})",
|
||||
incoming.peer_id, incoming.set_id);
|
||||
self.peerset.dropped(
|
||||
incoming.set_id,
|
||||
incoming.peer_id,
|
||||
sc_peerset::DropReason::Unknown,
|
||||
);
|
||||
self.peerset.dropped(incoming.set_id, incoming.peer_id, DropReason::Unknown);
|
||||
},
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
let state = match self.peers.get_mut(&(incoming.peer_id.clone(), incoming.set_id)) {
|
||||
let state = match self.peers.get_mut(&(incoming.peer_id, incoming.set_id)) {
|
||||
Some(s) => s,
|
||||
None => {
|
||||
debug_assert!(false);
|
||||
@@ -993,7 +999,7 @@ impl Notifications {
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
|
||||
incoming.peer_id, *connec_id, incoming.set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: incoming.peer_id.clone(),
|
||||
peer_id: incoming.peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Open { protocol_index: incoming.set_id.into() },
|
||||
});
|
||||
@@ -1029,7 +1035,7 @@ impl Notifications {
|
||||
return
|
||||
}
|
||||
|
||||
let state = match self.peers.get_mut(&(incoming.peer_id.clone(), incoming.set_id)) {
|
||||
let state = match self.peers.get_mut(&(incoming.peer_id, incoming.set_id)) {
|
||||
Some(s) => s,
|
||||
None => {
|
||||
debug_assert!(false);
|
||||
@@ -1053,7 +1059,7 @@ impl Notifications {
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})",
|
||||
incoming.peer_id, connec_id, incoming.set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: incoming.peer_id.clone(),
|
||||
peer_id: incoming.peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Close { protocol_index: incoming.set_id.into() },
|
||||
});
|
||||
@@ -1090,7 +1096,7 @@ impl NetworkBehaviour for Notifications {
|
||||
endpoint: &ConnectedPoint,
|
||||
) {
|
||||
for set_id in (0..self.notif_protocols.len()).map(sc_peerset::SetId::from) {
|
||||
match self.peers.entry((peer_id.clone(), set_id)).or_insert(PeerState::Poisoned) {
|
||||
match self.peers.entry((*peer_id, set_id)).or_insert(PeerState::Poisoned) {
|
||||
// Requested | PendingRequest => Enabled
|
||||
st @ &mut PeerState::Requested | st @ &mut PeerState::PendingRequest { .. } => {
|
||||
trace!(target: "sub-libp2p",
|
||||
@@ -1099,7 +1105,7 @@ impl NetworkBehaviour for Notifications {
|
||||
);
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})", peer_id, *conn, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
handler: NotifyHandler::One(*conn),
|
||||
event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -1148,9 +1154,7 @@ impl NetworkBehaviour for Notifications {
|
||||
_endpoint: &ConnectedPoint,
|
||||
) {
|
||||
for set_id in (0..self.notif_protocols.len()).map(sc_peerset::SetId::from) {
|
||||
let mut entry = if let Entry::Occupied(entry) =
|
||||
self.peers.entry((peer_id.clone(), set_id))
|
||||
{
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((*peer_id, set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "inject_connection_closed: State mismatch in the custom protos handler");
|
||||
@@ -1179,7 +1183,7 @@ impl NetworkBehaviour for Notifications {
|
||||
let delay_id = self.next_delay_id;
|
||||
self.next_delay_id.0 += 1;
|
||||
let delay = futures_timer::Delay::new(until - now);
|
||||
let peer_id = peer_id.clone();
|
||||
let peer_id = *peer_id;
|
||||
self.delays.push(
|
||||
async move {
|
||||
delay.await;
|
||||
@@ -1219,11 +1223,7 @@ impl NetworkBehaviour for Notifications {
|
||||
|
||||
if connections.is_empty() {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
peer_id.clone(),
|
||||
sc_peerset::DropReason::Unknown,
|
||||
);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
*entry.get_mut() = PeerState::Backoff { timer, timer_deadline };
|
||||
} else {
|
||||
*entry.get_mut() =
|
||||
@@ -1246,9 +1246,9 @@ impl NetworkBehaviour for Notifications {
|
||||
if let Some(pos) = connections.iter().position(|(c, _)| *c == *conn) {
|
||||
connections.remove(pos);
|
||||
} else {
|
||||
debug_assert!(false);
|
||||
error!(target: "sub-libp2p",
|
||||
"inject_connection_closed: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
}
|
||||
|
||||
let no_desired_left = !connections
|
||||
@@ -1280,7 +1280,7 @@ impl NetworkBehaviour for Notifications {
|
||||
let delay_id = self.next_delay_id;
|
||||
self.next_delay_id.0 += 1;
|
||||
let delay = futures_timer::Delay::new(until - now);
|
||||
let peer_id = peer_id.clone();
|
||||
let peer_id = *peer_id;
|
||||
self.delays.push(
|
||||
async move {
|
||||
delay.await;
|
||||
@@ -1322,15 +1322,11 @@ impl NetworkBehaviour for Notifications {
|
||||
if let Some(pos) = connections.iter().position(|(c, _)| *c == *conn) {
|
||||
let (_, state) = connections.remove(pos);
|
||||
if let ConnectionState::Open(_) = state {
|
||||
if let Some((replacement_pos, replacement_sink)) = connections
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(num, (_, s))| match s {
|
||||
if let Some((replacement_pos, replacement_sink)) =
|
||||
connections.iter().enumerate().find_map(|(num, (_, s))| match s {
|
||||
ConnectionState::Open(s) => Some((num, s.clone())),
|
||||
_ => None,
|
||||
})
|
||||
.next()
|
||||
{
|
||||
}) {
|
||||
if pos <= replacement_pos {
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
@@ -1338,7 +1334,7 @@ impl NetworkBehaviour for Notifications {
|
||||
peer_id, set_id
|
||||
);
|
||||
let event = NotificationsOut::CustomProtocolReplaced {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
set_id,
|
||||
notifications_sink: replacement_sink,
|
||||
};
|
||||
@@ -1351,7 +1347,7 @@ impl NetworkBehaviour for Notifications {
|
||||
peer_id, set_id
|
||||
);
|
||||
let event = NotificationsOut::CustomProtocolClosed {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
set_id,
|
||||
};
|
||||
self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
|
||||
@@ -1365,17 +1361,13 @@ impl NetworkBehaviour for Notifications {
|
||||
|
||||
if connections.is_empty() {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
peer_id.clone(),
|
||||
sc_peerset::DropReason::Unknown,
|
||||
);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
|
||||
|
||||
let delay_id = self.next_delay_id;
|
||||
self.next_delay_id.0 += 1;
|
||||
let delay = futures_timer::Delay::new(Duration::from_secs(ban_dur));
|
||||
let peer_id = peer_id.clone();
|
||||
let peer_id = *peer_id;
|
||||
self.delays.push(
|
||||
async move {
|
||||
delay.await;
|
||||
@@ -1392,11 +1384,7 @@ impl NetworkBehaviour for Notifications {
|
||||
matches!(s, ConnectionState::Opening | ConnectionState::Open(_))
|
||||
}) {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
peer_id.clone(),
|
||||
sc_peerset::DropReason::Unknown,
|
||||
);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
|
||||
*entry.get_mut() = PeerState::Disabled { connections, backoff_until: None };
|
||||
} else {
|
||||
@@ -1446,11 +1434,7 @@ impl NetworkBehaviour for Notifications {
|
||||
// "Basic" situation: we failed to reach a peer that the peerset requested.
|
||||
st @ PeerState::Requested | st @ PeerState::PendingRequest { .. } => {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
peer_id.clone(),
|
||||
sc_peerset::DropReason::Unknown,
|
||||
);
|
||||
self.peerset.dropped(set_id, *peer_id, DropReason::Unknown);
|
||||
|
||||
let now = Instant::now();
|
||||
let ban_duration = match st {
|
||||
@@ -1463,7 +1447,7 @@ impl NetworkBehaviour for Notifications {
|
||||
let delay_id = self.next_delay_id;
|
||||
self.next_delay_id.0 += 1;
|
||||
let delay = futures_timer::Delay::new(ban_duration);
|
||||
let peer_id = peer_id.clone();
|
||||
let peer_id = *peer_id;
|
||||
self.delays.push(
|
||||
async move {
|
||||
delay.await;
|
||||
@@ -1505,14 +1489,16 @@ impl NetworkBehaviour for Notifications {
|
||||
"Handler({:?}, {:?}]) => OpenDesiredByRemote({:?})",
|
||||
source, connection, set_id);
|
||||
|
||||
let mut entry =
|
||||
if let Entry::Occupied(entry) = self.peers.entry((source.clone(), set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "OpenDesiredByRemote: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((source, set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"OpenDesiredByRemote: State mismatch in the custom protos handler"
|
||||
);
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
|
||||
match mem::replace(entry.get_mut(), PeerState::Poisoned) {
|
||||
// Incoming => Incoming
|
||||
@@ -1601,9 +1587,9 @@ impl NetworkBehaviour for Notifications {
|
||||
|
||||
trace!(target: "sub-libp2p", "PSM <= Incoming({}, {:?}).",
|
||||
source, incoming_id);
|
||||
self.peerset.incoming(set_id, source.clone(), incoming_id);
|
||||
self.peerset.incoming(set_id, source, incoming_id);
|
||||
self.incoming.push(IncomingPeer {
|
||||
peer_id: source.clone(),
|
||||
peer_id: source,
|
||||
set_id,
|
||||
alive: true,
|
||||
incoming_id,
|
||||
@@ -1641,7 +1627,7 @@ impl NetworkBehaviour for Notifications {
|
||||
trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
|
||||
source, connection, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: source.clone(),
|
||||
peer_id: source,
|
||||
handler: NotifyHandler::One(connection),
|
||||
event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
|
||||
});
|
||||
@@ -1689,14 +1675,13 @@ impl NetworkBehaviour for Notifications {
|
||||
"Handler({}, {:?}) => CloseDesired({:?})",
|
||||
source, connection, set_id);
|
||||
|
||||
let mut entry =
|
||||
if let Entry::Occupied(entry) = self.peers.entry((source.clone(), set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "CloseDesired: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((source, set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "CloseDesired: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
|
||||
match mem::replace(entry.get_mut(), PeerState::Poisoned) {
|
||||
// Enabled => Enabled | Disabled
|
||||
@@ -1727,20 +1712,16 @@ impl NetworkBehaviour for Notifications {
|
||||
|
||||
trace!(target: "sub-libp2p", "Handler({}, {:?}) <= Close({:?})", source, connection, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: source.clone(),
|
||||
peer_id: source,
|
||||
handler: NotifyHandler::One(connection),
|
||||
event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
|
||||
});
|
||||
|
||||
if let Some((replacement_pos, replacement_sink)) = connections
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(num, (_, s))| match s {
|
||||
if let Some((replacement_pos, replacement_sink)) =
|
||||
connections.iter().enumerate().find_map(|(num, (_, s))| match s {
|
||||
ConnectionState::Open(s) => Some((num, s.clone())),
|
||||
_ => None,
|
||||
})
|
||||
.next()
|
||||
{
|
||||
}) {
|
||||
if pos <= replacement_pos {
|
||||
trace!(target: "sub-libp2p", "External API <= Sink replaced({:?})", source);
|
||||
let event = NotificationsOut::CustomProtocolReplaced {
|
||||
@@ -1759,11 +1740,7 @@ impl NetworkBehaviour for Notifications {
|
||||
.any(|(_, s)| matches!(s, ConnectionState::Opening))
|
||||
{
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", source, set_id);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
source.clone(),
|
||||
sc_peerset::DropReason::Refused,
|
||||
);
|
||||
self.peerset.dropped(set_id, source, DropReason::Refused);
|
||||
*entry.into_mut() =
|
||||
PeerState::Disabled { connections, backoff_until: None };
|
||||
} else {
|
||||
@@ -1838,7 +1815,7 @@ impl NetworkBehaviour for Notifications {
|
||||
"Handler({}, {:?}) => OpenResultOk({:?})",
|
||||
source, connection, set_id);
|
||||
|
||||
match self.peers.get_mut(&(source.clone(), set_id)) {
|
||||
match self.peers.get_mut(&(source, set_id)) {
|
||||
Some(PeerState::Enabled { connections, .. }) => {
|
||||
debug_assert!(connections.iter().any(|(_, s)| matches!(
|
||||
s,
|
||||
@@ -1868,9 +1845,9 @@ impl NetworkBehaviour for Notifications {
|
||||
}) {
|
||||
*connec_state = ConnectionState::Closing;
|
||||
} else {
|
||||
debug_assert!(false);
|
||||
error!(target: "sub-libp2p",
|
||||
"OpenResultOk State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
}
|
||||
},
|
||||
|
||||
@@ -1904,15 +1881,13 @@ impl NetworkBehaviour for Notifications {
|
||||
"Handler({:?}, {:?}) => OpenResultErr({:?})",
|
||||
source, connection, set_id);
|
||||
|
||||
let mut entry =
|
||||
if let Entry::Occupied(entry) = self.peers.entry((source.clone(), set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "OpenResultErr: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
let mut entry = if let Entry::Occupied(entry) = self.peers.entry((source, set_id)) {
|
||||
entry
|
||||
} else {
|
||||
error!(target: "sub-libp2p", "OpenResultErr: State mismatch in the custom protos handler");
|
||||
debug_assert!(false);
|
||||
return
|
||||
};
|
||||
|
||||
match mem::replace(entry.get_mut(), PeerState::Poisoned) {
|
||||
PeerState::Enabled { mut connections } => {
|
||||
@@ -1940,11 +1915,7 @@ impl NetworkBehaviour for Notifications {
|
||||
matches!(s, ConnectionState::Opening | ConnectionState::Open(_))
|
||||
}) {
|
||||
trace!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
|
||||
self.peerset.dropped(
|
||||
set_id,
|
||||
source.clone(),
|
||||
sc_peerset::DropReason::Refused,
|
||||
);
|
||||
self.peerset.dropped(set_id, source, DropReason::Refused);
|
||||
|
||||
let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
|
||||
*entry.into_mut() = PeerState::Disabled {
|
||||
@@ -2002,8 +1973,12 @@ impl NetworkBehaviour for Notifications {
|
||||
set_id,
|
||||
message.len()
|
||||
);
|
||||
trace!(target: "sub-libp2p", "External API <= Message({}, {:?})",
|
||||
source, set_id);
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"External API <= Message({}, {:?})",
|
||||
source,
|
||||
set_id,
|
||||
);
|
||||
let event = NotificationsOut::Notification { peer_id: source, set_id, message };
|
||||
|
||||
self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
|
||||
@@ -2057,7 +2032,7 @@ impl NetworkBehaviour for Notifications {
|
||||
while let Poll::Ready(Some((delay_id, peer_id, set_id))) =
|
||||
Pin::new(&mut self.delays).poll_next(cx)
|
||||
{
|
||||
let peer_state = match self.peers.get_mut(&(peer_id.clone(), set_id)) {
|
||||
let peer_state = match self.peers.get_mut(&(peer_id, set_id)) {
|
||||
Some(s) => s,
|
||||
// We intentionally never remove elements from `delays`, and it may
|
||||
// thus contain peers which are now gone. This is a normal situation.
|
||||
@@ -2090,7 +2065,7 @@ impl NetworkBehaviour for Notifications {
|
||||
trace!(target: "sub-libp2p", "Handler({}, {:?}) <= Open({:?}) (ban expired)",
|
||||
peer_id, *connec_id, set_id);
|
||||
self.events.push_back(NetworkBehaviourAction::NotifyHandler {
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id,
|
||||
handler: NotifyHandler::One(*connec_id),
|
||||
event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
|
||||
});
|
||||
|
||||
@@ -256,7 +256,7 @@ impl IntoProtocolsHandler for NotifsHandlerProto {
|
||||
Protocol { config, in_upgrade, state: State::Closed { pending_opening: false } }
|
||||
})
|
||||
.collect(),
|
||||
peer_id: peer_id.clone(),
|
||||
peer_id: *peer_id,
|
||||
endpoint: connected_point.clone(),
|
||||
when_connection_open: Instant::now(),
|
||||
events_queue: VecDeque::with_capacity(16),
|
||||
@@ -463,7 +463,7 @@ impl NotifsHandlerProto {
|
||||
/// is always the same whether we open a substream ourselves or respond to handshake from
|
||||
/// the remote.
|
||||
pub fn new(list: impl Into<Vec<ProtocolConfig>>) -> Self {
|
||||
NotifsHandlerProto { protocols: list.into() }
|
||||
Self { protocols: list.into() }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -552,7 +552,7 @@ impl ProtocolsHandler for NotifsHandler {
|
||||
let (sync_tx, sync_rx) = mpsc::channel(SYNC_NOTIFICATIONS_BUFFER_SIZE);
|
||||
let notifications_sink = NotificationsSink {
|
||||
inner: Arc::new(NotificationsSinkInner {
|
||||
peer_id: self.peer_id.clone(),
|
||||
peer_id: self.peer_id,
|
||||
async_channel: FuturesMutex::new(async_tx),
|
||||
sync_channel: Mutex::new(sync_tx),
|
||||
}),
|
||||
|
||||
@@ -34,13 +34,13 @@ pub struct UpgradeCollec<T>(pub Vec<T>);
|
||||
|
||||
impl<T> From<Vec<T>> for UpgradeCollec<T> {
|
||||
fn from(list: Vec<T>) -> Self {
|
||||
UpgradeCollec(list)
|
||||
Self(list)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> FromIterator<T> for UpgradeCollec<T> {
|
||||
fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
|
||||
UpgradeCollec(iter.into_iter().collect())
|
||||
Self(iter.into_iter().collect())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -38,7 +38,7 @@ use asynchronous_codec::Framed;
|
||||
use bytes::BytesMut;
|
||||
use futures::prelude::*;
|
||||
use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
|
||||
use log::error;
|
||||
use log::{error, warn};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
convert::{Infallible, TryFrom as _},
|
||||
@@ -121,7 +121,7 @@ impl NotificationsIn {
|
||||
let mut protocol_names = fallback_names;
|
||||
protocol_names.insert(0, main_protocol_name.into());
|
||||
|
||||
NotificationsIn { protocol_names, max_notification_size }
|
||||
Self { protocol_names, max_notification_size }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -347,7 +347,7 @@ impl NotificationsOut {
|
||||
let mut protocol_names = fallback_names;
|
||||
protocol_names.insert(0, main_protocol_name.into());
|
||||
|
||||
NotificationsOut { protocol_names, initial_message, max_notification_size }
|
||||
Self { protocol_names, initial_message, max_notification_size }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -478,12 +478,11 @@ pub enum NotificationsHandshakeError {
|
||||
impl From<unsigned_varint::io::ReadError> for NotificationsHandshakeError {
|
||||
fn from(err: unsigned_varint::io::ReadError) -> Self {
|
||||
match err {
|
||||
unsigned_varint::io::ReadError::Io(err) => NotificationsHandshakeError::Io(err),
|
||||
unsigned_varint::io::ReadError::Decode(err) =>
|
||||
NotificationsHandshakeError::VarintDecode(err),
|
||||
unsigned_varint::io::ReadError::Io(err) => Self::Io(err),
|
||||
unsigned_varint::io::ReadError::Decode(err) => Self::VarintDecode(err),
|
||||
_ => {
|
||||
log::warn!("Unrecognized varint decoding error");
|
||||
NotificationsHandshakeError::Io(From::from(io::ErrorKind::InvalidData))
|
||||
warn!("Unrecognized varint decoding error");
|
||||
Self::Io(From::from(io::ErrorKind::InvalidData))
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,40 +148,37 @@ enum PendingRequests {
|
||||
|
||||
impl PendingRequests {
|
||||
fn add(&mut self, id: &PeerId) {
|
||||
match self {
|
||||
PendingRequests::Some(set) => {
|
||||
set.insert(id.clone());
|
||||
},
|
||||
PendingRequests::All => {},
|
||||
if let Self::Some(ref mut set) = self {
|
||||
set.insert(*id);
|
||||
}
|
||||
}
|
||||
|
||||
fn take(&mut self) -> PendingRequests {
|
||||
fn take(&mut self) -> Self {
|
||||
std::mem::take(self)
|
||||
}
|
||||
|
||||
fn set_all(&mut self) {
|
||||
*self = PendingRequests::All;
|
||||
*self = Self::All;
|
||||
}
|
||||
|
||||
fn contains(&self, id: &PeerId) -> bool {
|
||||
match self {
|
||||
PendingRequests::Some(set) => set.contains(id),
|
||||
PendingRequests::All => true,
|
||||
Self::Some(set) => set.contains(id),
|
||||
Self::All => true,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_empty(&self) -> bool {
|
||||
match self {
|
||||
PendingRequests::Some(set) => set.is_empty(),
|
||||
PendingRequests::All => false,
|
||||
Self::Some(set) => set.is_empty(),
|
||||
Self::All => false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for PendingRequests {
|
||||
fn default() -> Self {
|
||||
PendingRequests::Some(HashSet::default())
|
||||
Self::Some(HashSet::default())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -343,10 +340,10 @@ pub enum WarpSyncPhase {
|
||||
impl fmt::Display for WarpSyncPhase {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
match self {
|
||||
WarpSyncPhase::AwaitingPeers => write!(f, "Waiting for peers"),
|
||||
WarpSyncPhase::DownloadingWarpProofs => write!(f, "Downloading finality proofs"),
|
||||
WarpSyncPhase::DownloadingState => write!(f, "Downloading state"),
|
||||
WarpSyncPhase::ImportingState => write!(f, "Importing state"),
|
||||
Self::AwaitingPeers => write!(f, "Waiting for peers"),
|
||||
Self::DownloadingWarpProofs => write!(f, "Downloading finality proofs"),
|
||||
Self::DownloadingState => write!(f, "Downloading state"),
|
||||
Self::ImportingState => write!(f, "Importing state"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -538,7 +535,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
max_parallel_downloads: u32,
|
||||
warp_sync_provider: Option<Arc<dyn WarpSyncProvider<B>>>,
|
||||
) -> Result<Self, ClientError> {
|
||||
let mut sync = ChainSync {
|
||||
let mut sync = Self {
|
||||
client,
|
||||
peers: HashMap::new(),
|
||||
blocks: BlockCollection::new(),
|
||||
@@ -732,7 +729,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
|
||||
self.pending_requests.add(&who);
|
||||
self.peers.insert(
|
||||
who.clone(),
|
||||
who,
|
||||
PeerSync {
|
||||
peer_id: who,
|
||||
common_number: Zero::zero(),
|
||||
@@ -754,9 +751,9 @@ impl<B: BlockT> ChainSync<B> {
|
||||
best_number,
|
||||
);
|
||||
self.peers.insert(
|
||||
who.clone(),
|
||||
who,
|
||||
PeerSync {
|
||||
peer_id: who.clone(),
|
||||
peer_id: who,
|
||||
common_number: std::cmp::min(self.best_queued_number, best_number),
|
||||
best_hash,
|
||||
best_number,
|
||||
@@ -835,7 +832,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
}
|
||||
|
||||
self.fork_targets
|
||||
.entry(hash.clone())
|
||||
.entry(*hash)
|
||||
.or_insert_with(|| ForkTarget { number, peers: Default::default(), parent_hash: None })
|
||||
.peers
|
||||
.extend(peers);
|
||||
@@ -972,7 +969,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
trace!(target: "sync", "New StateRequest for {}", id);
|
||||
peer.state = PeerSyncState::DownloadingState;
|
||||
let request = sync.next_request();
|
||||
return Some((id.clone(), request))
|
||||
return Some((*id, request))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -987,7 +984,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
if peer.state.is_available() && peer.best_number >= target {
|
||||
trace!(target: "sync", "New StateRequest for {}", id);
|
||||
peer.state = PeerSyncState::DownloadingState;
|
||||
return Some((id.clone(), request))
|
||||
return Some((*id, request))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1019,7 +1016,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
if peer.state.is_available() && peer.best_number >= median {
|
||||
trace!(target: "sync", "New WarpProofRequest for {}", id);
|
||||
peer.state = PeerSyncState::DownloadingWarpProof;
|
||||
return Some((id.clone(), request))
|
||||
return Some((*id, request))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1068,7 +1065,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
peer.state = PeerSyncState::Available;
|
||||
if blocks.is_empty() {
|
||||
debug!(target: "sync", "Empty block response from {}", who);
|
||||
return Err(BadPeer(who.clone(), rep::NO_BLOCK))
|
||||
return Err(BadPeer(*who, rep::NO_BLOCK))
|
||||
}
|
||||
validate_blocks::<B>(&blocks, who, Some(request))?;
|
||||
blocks
|
||||
@@ -1083,7 +1080,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
body: b.body,
|
||||
indexed_body: None,
|
||||
justifications,
|
||||
origin: Some(who.clone()),
|
||||
origin: Some(*who),
|
||||
allow_missing_state: true,
|
||||
import_existing: self.import_existing,
|
||||
skip_execution: self.skip_execution(),
|
||||
@@ -1110,7 +1107,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
"Invalid response when searching for ancestor from {}",
|
||||
who,
|
||||
);
|
||||
return Err(BadPeer(who.clone(), rep::UNKNOWN_ANCESTOR))
|
||||
return Err(BadPeer(*who, rep::UNKNOWN_ANCESTOR))
|
||||
},
|
||||
(_, Err(e)) => {
|
||||
info!(
|
||||
@@ -1118,7 +1115,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
"❌ Error answering legitimate blockchain query: {:?}",
|
||||
e,
|
||||
);
|
||||
return Err(BadPeer(who.clone(), rep::BLOCKCHAIN_READ_ERROR))
|
||||
return Err(BadPeer(*who, rep::BLOCKCHAIN_READ_ERROR))
|
||||
},
|
||||
};
|
||||
if matching_hash.is_some() {
|
||||
@@ -1135,7 +1132,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
}
|
||||
if matching_hash.is_none() && current.is_zero() {
|
||||
trace!(target:"sync", "Ancestry search: genesis mismatch for peer {}", who);
|
||||
return Err(BadPeer(who.clone(), rep::GENESIS_MISMATCH))
|
||||
return Err(BadPeer(*who, rep::GENESIS_MISMATCH))
|
||||
}
|
||||
if let Some((next_state, next_num)) =
|
||||
handle_ancestor_search_state(state, *current, matching_hash.is_some())
|
||||
@@ -1145,10 +1142,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
start: *start,
|
||||
state: next_state,
|
||||
};
|
||||
return Ok(OnBlockData::Request(
|
||||
who.clone(),
|
||||
ancestry_request::<B>(next_num),
|
||||
))
|
||||
return Ok(OnBlockData::Request(*who, ancestry_request::<B>(next_num)))
|
||||
} else {
|
||||
// Ancestry search is complete. Check if peer is on a stale fork unknown
|
||||
// to us and add it to sync targets if necessary.
|
||||
@@ -1172,14 +1166,14 @@ impl<B: BlockT> ChainSync<B> {
|
||||
who,
|
||||
);
|
||||
self.fork_targets
|
||||
.entry(peer.best_hash.clone())
|
||||
.entry(peer.best_hash)
|
||||
.or_insert_with(|| ForkTarget {
|
||||
number: peer.best_number,
|
||||
parent_hash: None,
|
||||
peers: Default::default(),
|
||||
})
|
||||
.peers
|
||||
.insert(who.clone());
|
||||
.insert(*who);
|
||||
}
|
||||
peer.state = PeerSyncState::Available;
|
||||
Vec::new()
|
||||
@@ -1204,7 +1198,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
body: b.body,
|
||||
indexed_body: None,
|
||||
justifications,
|
||||
origin: Some(who.clone()),
|
||||
origin: Some(*who),
|
||||
allow_missing_state: true,
|
||||
import_existing: false,
|
||||
skip_execution: true,
|
||||
@@ -1215,7 +1209,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
}
|
||||
} else {
|
||||
// We don't know of this peer, so we also did not request anything from it.
|
||||
return Err(BadPeer(who.clone(), rep::NOT_REQUESTED))
|
||||
return Err(BadPeer(*who, rep::NOT_REQUESTED))
|
||||
};
|
||||
|
||||
Ok(self.validate_and_queue_blocks(new_blocks))
|
||||
@@ -1249,7 +1243,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
sync.import_state(response)
|
||||
} else {
|
||||
debug!(target: "sync", "Ignored obsolete state response from {}", who);
|
||||
return Err(BadPeer(who.clone(), rep::NOT_REQUESTED))
|
||||
return Err(BadPeer(*who, rep::NOT_REQUESTED))
|
||||
};
|
||||
|
||||
match import_result {
|
||||
@@ -1274,7 +1268,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
Ok(OnStateData::Request(who.clone(), request)),
|
||||
state::ImportResult::BadResponse => {
|
||||
debug!(target: "sync", "Bad state data received from {}", who);
|
||||
Err(BadPeer(who.clone(), rep::BAD_BLOCK))
|
||||
Err(BadPeer(*who, rep::BAD_BLOCK))
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1297,17 +1291,17 @@ impl<B: BlockT> ChainSync<B> {
|
||||
sync.import_warp_proof(response)
|
||||
} else {
|
||||
debug!(target: "sync", "Ignored obsolete warp sync response from {}", who);
|
||||
return Err(BadPeer(who.clone(), rep::NOT_REQUESTED))
|
||||
return Err(BadPeer(*who, rep::NOT_REQUESTED))
|
||||
};
|
||||
|
||||
match import_result {
|
||||
warp::WarpProofImportResult::StateRequest(request) =>
|
||||
Ok(OnWarpSyncData::StateRequest(who.clone(), request)),
|
||||
Ok(OnWarpSyncData::StateRequest(*who, request)),
|
||||
warp::WarpProofImportResult::WarpProofRequest(request) =>
|
||||
Ok(OnWarpSyncData::WarpProofRequest(who.clone(), request)),
|
||||
Ok(OnWarpSyncData::WarpProofRequest(*who, request)),
|
||||
warp::WarpProofImportResult::BadResponse => {
|
||||
debug!(target: "sync", "Bad proof data received from {}", who);
|
||||
Err(BadPeer(who.clone(), rep::BAD_BLOCK))
|
||||
Err(BadPeer(*who, rep::BAD_BLOCK))
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1319,7 +1313,11 @@ impl<B: BlockT> ChainSync<B> {
|
||||
let orig_len = new_blocks.len();
|
||||
new_blocks.retain(|b| !self.queue_blocks.contains(&b.hash));
|
||||
if new_blocks.len() != orig_len {
|
||||
debug!(target: "sync", "Ignoring {} blocks that are already queued", orig_len - new_blocks.len());
|
||||
debug!(
|
||||
target: "sync",
|
||||
"Ignoring {} blocks that are already queued",
|
||||
orig_len - new_blocks.len(),
|
||||
);
|
||||
}
|
||||
|
||||
let origin = if self.status().state != SyncState::Downloading {
|
||||
@@ -1372,7 +1370,10 @@ impl<B: BlockT> ChainSync<B> {
|
||||
if hash != block.hash {
|
||||
warn!(
|
||||
target: "sync",
|
||||
"💔 Invalid block justification provided by {}: requested: {:?} got: {:?}", who, hash, block.hash
|
||||
"💔 Invalid block justification provided by {}: requested: {:?} got: {:?}",
|
||||
who,
|
||||
hash,
|
||||
block.hash,
|
||||
);
|
||||
return Err(BadPeer(who, rep::BAD_JUSTIFICATION))
|
||||
}
|
||||
@@ -1381,7 +1382,8 @@ impl<B: BlockT> ChainSync<B> {
|
||||
} else {
|
||||
// we might have asked the peer for a justification on a block that we assumed it
|
||||
// had but didn't (regardless of whether it had a justification for it or not).
|
||||
trace!(target: "sync",
|
||||
trace!(
|
||||
target: "sync",
|
||||
"Peer {:?} provided empty response for justification request {:?}",
|
||||
who,
|
||||
hash,
|
||||
@@ -1441,7 +1443,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
target: "sync",
|
||||
"Block imported clears all pending justification requests {}: {:?}",
|
||||
number,
|
||||
hash
|
||||
hash,
|
||||
);
|
||||
self.clear_justification_requests();
|
||||
}
|
||||
@@ -1459,7 +1461,7 @@ impl<B: BlockT> ChainSync<B> {
|
||||
if aux.bad_justification {
|
||||
if let Some(ref peer) = who {
|
||||
warn!("💔 Sent block with bad justification to import");
|
||||
output.push(Err(BadPeer(peer.clone(), rep::BAD_JUSTIFICATION)));
|
||||
output.push(Err(BadPeer(*peer, rep::BAD_JUSTIFICATION)));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1934,14 +1936,14 @@ impl<B: BlockT> ChainSync<B> {
|
||||
announce.summary(),
|
||||
);
|
||||
self.fork_targets
|
||||
.entry(hash.clone())
|
||||
.entry(hash)
|
||||
.or_insert_with(|| ForkTarget {
|
||||
number,
|
||||
parent_hash: Some(*announce.header.parent_hash()),
|
||||
peers: Default::default(),
|
||||
})
|
||||
.peers
|
||||
.insert(who.clone());
|
||||
.insert(who);
|
||||
}
|
||||
|
||||
PollBlockAnnounceValidation::Nothing { is_best, who, announce }
|
||||
@@ -2008,14 +2010,14 @@ impl<B: BlockT> ChainSync<B> {
|
||||
fn reset_sync_start_point(&mut self) -> Result<(), ClientError> {
|
||||
let info = self.client.info();
|
||||
if matches!(self.mode, SyncMode::LightState { .. }) && info.finalized_state.is_some() {
|
||||
log::warn!(
|
||||
warn!(
|
||||
target: "sync",
|
||||
"Can't use fast sync mode with a partially synced database. Reverting to full sync mode."
|
||||
);
|
||||
self.mode = SyncMode::Full;
|
||||
}
|
||||
if matches!(self.mode, SyncMode::Warp) && info.finalized_state.is_some() {
|
||||
log::warn!(
|
||||
warn!(
|
||||
target: "sync",
|
||||
"Can't use warp sync mode with a partially synced database. Reverting to full sync mode."
|
||||
);
|
||||
@@ -2031,17 +2033,17 @@ impl<B: BlockT> ChainSync<B> {
|
||||
self.import_existing = true;
|
||||
// Latest state is missing, start with the last finalized state or genesis instead.
|
||||
if let Some((hash, number)) = info.finalized_state {
|
||||
log::debug!(target: "sync", "Starting from finalized state #{}", number);
|
||||
debug!(target: "sync", "Starting from finalized state #{}", number);
|
||||
self.best_queued_hash = hash;
|
||||
self.best_queued_number = number;
|
||||
} else {
|
||||
log::debug!(target: "sync", "Restarting from genesis");
|
||||
debug!(target: "sync", "Restarting from genesis");
|
||||
self.best_queued_hash = Default::default();
|
||||
self.best_queued_number = Zero::zero();
|
||||
}
|
||||
}
|
||||
}
|
||||
log::trace!(target: "sync", "Restarted sync at #{} ({:?})", self.best_queued_number, self.best_queued_hash);
|
||||
trace!(target: "sync", "Restarted sync at #{} ({:?})", self.best_queued_number, self.best_queued_hash);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -2219,7 +2221,7 @@ fn peer_block_request<B: BlockT>(
|
||||
);
|
||||
}
|
||||
let range = blocks.needed_blocks(
|
||||
id.clone(),
|
||||
*id,
|
||||
MAX_BLOCKS_TO_REQUEST,
|
||||
peer.best_number,
|
||||
peer.common_number,
|
||||
@@ -2335,7 +2337,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
blocks.len(),
|
||||
);
|
||||
|
||||
return Err(BadPeer(who.clone(), rep::NOT_REQUESTED))
|
||||
return Err(BadPeer(*who, rep::NOT_REQUESTED))
|
||||
}
|
||||
|
||||
let block_header = if request.direction == message::Direction::Descending {
|
||||
@@ -2358,7 +2360,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
block_header,
|
||||
);
|
||||
|
||||
return Err(BadPeer(who.clone(), rep::NOT_REQUESTED))
|
||||
return Err(BadPeer(*who, rep::NOT_REQUESTED))
|
||||
}
|
||||
|
||||
if request.fields.contains(message::BlockAttributes::HEADER) &&
|
||||
@@ -2370,7 +2372,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
who,
|
||||
);
|
||||
|
||||
return Err(BadPeer(who.clone(), rep::BAD_RESPONSE))
|
||||
return Err(BadPeer(*who, rep::BAD_RESPONSE))
|
||||
}
|
||||
|
||||
if request.fields.contains(message::BlockAttributes::BODY) &&
|
||||
@@ -2382,7 +2384,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
who,
|
||||
);
|
||||
|
||||
return Err(BadPeer(who.clone(), rep::BAD_RESPONSE))
|
||||
return Err(BadPeer(*who, rep::BAD_RESPONSE))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2397,7 +2399,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
b.hash,
|
||||
hash,
|
||||
);
|
||||
return Err(BadPeer(who.clone(), rep::BAD_BLOCK))
|
||||
return Err(BadPeer(*who, rep::BAD_BLOCK))
|
||||
}
|
||||
}
|
||||
if let (Some(header), Some(body)) = (&b.header, &b.body) {
|
||||
@@ -2413,7 +2415,7 @@ fn validate_blocks<Block: BlockT>(
|
||||
expected,
|
||||
got,
|
||||
);
|
||||
return Err(BadPeer(who.clone(), rep::BAD_BLOCK))
|
||||
return Err(BadPeer(*who, rep::BAD_BLOCK))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2457,7 +2459,7 @@ mod test {
|
||||
};
|
||||
|
||||
// add a new peer with the same best block
|
||||
sync.new_peer(peer_id.clone(), a1_hash, a1_number).unwrap();
|
||||
sync.new_peer(peer_id, a1_hash, a1_number).unwrap();
|
||||
|
||||
// and request a justification for the block
|
||||
sync.request_justification(&a1_hash, a1_number);
|
||||
@@ -2478,10 +2480,7 @@ mod test {
|
||||
// if the peer replies with an empty response (i.e. it doesn't know the block),
|
||||
// the active request should be cleared.
|
||||
assert_eq!(
|
||||
sync.on_block_justification(
|
||||
peer_id.clone(),
|
||||
BlockResponse::<Block> { id: 0, blocks: vec![] }
|
||||
),
|
||||
sync.on_block_justification(peer_id, BlockResponse::<Block> { id: 0, blocks: vec![] }),
|
||||
Ok(OnBlockJustification::Nothing),
|
||||
);
|
||||
|
||||
|
||||
@@ -44,8 +44,8 @@ enum BlockRangeState<B: BlockT> {
|
||||
impl<B: BlockT> BlockRangeState<B> {
|
||||
pub fn len(&self) -> NumberFor<B> {
|
||||
match *self {
|
||||
BlockRangeState::Downloading { len, .. } => len,
|
||||
BlockRangeState::Complete(ref blocks) => (blocks.len() as u32).into(),
|
||||
Self::Downloading { len, .. } => len,
|
||||
Self::Complete(ref blocks) => (blocks.len() as u32).into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -61,7 +61,7 @@ pub struct BlockCollection<B: BlockT> {
|
||||
impl<B: BlockT> BlockCollection<B> {
|
||||
/// Create a new instance.
|
||||
pub fn new() -> Self {
|
||||
BlockCollection { blocks: BTreeMap::new(), peer_requests: HashMap::new() }
|
||||
Self { blocks: BTreeMap::new(), peer_requests: HashMap::new() }
|
||||
}
|
||||
|
||||
/// Clear everything.
|
||||
@@ -90,10 +90,7 @@ impl<B: BlockT> BlockCollection<B> {
|
||||
self.blocks.insert(
|
||||
start,
|
||||
BlockRangeState::Complete(
|
||||
blocks
|
||||
.into_iter()
|
||||
.map(|b| BlockData { origin: Some(who.clone()), block: b })
|
||||
.collect(),
|
||||
blocks.into_iter().map(|b| BlockData { origin: Some(who), block: b }).collect(),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ pub(crate) struct Metrics {
|
||||
|
||||
impl<B: BlockT> ExtraRequests<B> {
|
||||
pub(crate) fn new(request_type_name: &'static str) -> Self {
|
||||
ExtraRequests {
|
||||
Self {
|
||||
tree: ForkTree::new(),
|
||||
best_seen_finalized_number: Zero::zero(),
|
||||
pending_requests: VecDeque::new(),
|
||||
@@ -132,27 +132,25 @@ impl<B: BlockT> ExtraRequests<B> {
|
||||
// messages to chain sync.
|
||||
if let Some(request) = self.active_requests.remove(&who) {
|
||||
if let Some(r) = resp {
|
||||
trace!(target: "sync", "Queuing import of {} from {:?} for {:?}",
|
||||
self.request_type_name,
|
||||
who,
|
||||
request,
|
||||
trace!(target: "sync",
|
||||
"Queuing import of {} from {:?} for {:?}",
|
||||
self.request_type_name, who, request,
|
||||
);
|
||||
|
||||
self.importing_requests.insert(request);
|
||||
return Some((who, request.0, request.1, r))
|
||||
} else {
|
||||
trace!(target: "sync", "Empty {} response from {:?} for {:?}",
|
||||
self.request_type_name,
|
||||
who,
|
||||
request,
|
||||
trace!(target: "sync",
|
||||
"Empty {} response from {:?} for {:?}",
|
||||
self.request_type_name, who, request,
|
||||
);
|
||||
}
|
||||
self.failed_requests.entry(request).or_default().push((who, Instant::now()));
|
||||
self.pending_requests.push_front(request);
|
||||
} else {
|
||||
trace!(target: "sync", "No active {} request to {:?}",
|
||||
self.request_type_name,
|
||||
who,
|
||||
trace!(target: "sync",
|
||||
"No active {} request to {:?}",
|
||||
self.request_type_name, who,
|
||||
);
|
||||
}
|
||||
None
|
||||
@@ -227,10 +225,9 @@ impl<B: BlockT> ExtraRequests<B> {
|
||||
};
|
||||
|
||||
if self.tree.finalize_root(&finalized_hash).is_none() {
|
||||
warn!(target: "sync", "‼️ Imported {:?} {:?} which isn't a root in the tree: {:?}",
|
||||
finalized_hash,
|
||||
finalized_number,
|
||||
self.tree.roots().collect::<Vec<_>>()
|
||||
warn!(target: "sync",
|
||||
"‼️ Imported {:?} {:?} which isn't a root in the tree: {:?}",
|
||||
finalized_hash, finalized_number, self.tree.roots().collect::<Vec<_>>()
|
||||
);
|
||||
return true
|
||||
}
|
||||
@@ -280,7 +277,7 @@ pub(crate) struct Matcher<'a, B: BlockT> {
|
||||
|
||||
impl<'a, B: BlockT> Matcher<'a, B> {
|
||||
fn new(extras: &'a mut ExtraRequests<B>) -> Self {
|
||||
Matcher { remaining: extras.pending_requests.len(), extras }
|
||||
Self { remaining: extras.pending_requests.len(), extras }
|
||||
}
|
||||
|
||||
/// Finds a peer to which a pending request can be sent.
|
||||
@@ -335,13 +332,12 @@ impl<'a, B: BlockT> Matcher<'a, B> {
|
||||
}
|
||||
self.extras.active_requests.insert(peer.clone(), request);
|
||||
|
||||
trace!(target: "sync", "Sending {} request to {:?} for {:?}",
|
||||
self.extras.request_type_name,
|
||||
peer,
|
||||
request,
|
||||
trace!(target: "sync",
|
||||
"Sending {} request to {:?} for {:?}",
|
||||
self.extras.request_type_name, peer, request,
|
||||
);
|
||||
|
||||
return Some((peer.clone(), request))
|
||||
return Some((*peer, request))
|
||||
}
|
||||
|
||||
self.extras.pending_requests.push_back(request);
|
||||
@@ -594,7 +590,7 @@ mod tests {
|
||||
let mut peers = HashMap::with_capacity(g.size());
|
||||
for _ in 0..g.size() {
|
||||
let ps = ArbitraryPeerSync::arbitrary(g).0;
|
||||
peers.insert(ps.peer_id.clone(), ps);
|
||||
peers.insert(ps.peer_id, ps);
|
||||
}
|
||||
ArbitraryPeers(peers)
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ use crate::{
|
||||
schema::v1::{StateEntry, StateRequest, StateResponse},
|
||||
};
|
||||
use codec::{Decode, Encode};
|
||||
use log::debug;
|
||||
use sc_client_api::StorageProof;
|
||||
use sp_runtime::traits::{Block as BlockT, Header, NumberFor};
|
||||
use std::sync::Arc;
|
||||
@@ -55,7 +56,7 @@ pub enum ImportResult<B: BlockT> {
|
||||
impl<B: BlockT> StateSync<B> {
|
||||
/// Create a new instance.
|
||||
pub fn new(client: Arc<dyn Client<B>>, target: B::Header, skip_proof: bool) -> Self {
|
||||
StateSync {
|
||||
Self {
|
||||
client,
|
||||
target_block: target.hash(),
|
||||
target_root: target.state_root().clone(),
|
||||
@@ -71,46 +72,32 @@ impl<B: BlockT> StateSync<B> {
|
||||
/// Validate and import a state reponse.
|
||||
pub fn import(&mut self, response: StateResponse) -> ImportResult<B> {
|
||||
if response.entries.is_empty() && response.proof.is_empty() && !response.complete {
|
||||
log::debug!(
|
||||
target: "sync",
|
||||
"Bad state response",
|
||||
);
|
||||
debug!(target: "sync", "Bad state response");
|
||||
return ImportResult::BadResponse
|
||||
}
|
||||
if !self.skip_proof && response.proof.is_empty() {
|
||||
log::debug!(
|
||||
target: "sync",
|
||||
"Missing proof",
|
||||
);
|
||||
debug!(target: "sync", "Missing proof");
|
||||
return ImportResult::BadResponse
|
||||
}
|
||||
let complete = if !self.skip_proof {
|
||||
log::debug!(
|
||||
target: "sync",
|
||||
"Importing state from {} trie nodes",
|
||||
response.proof.len(),
|
||||
);
|
||||
debug!(target: "sync", "Importing state from {} trie nodes", response.proof.len());
|
||||
let proof_size = response.proof.len() as u64;
|
||||
let proof = match StorageProof::decode(&mut response.proof.as_ref()) {
|
||||
Ok(proof) => proof,
|
||||
Err(e) => {
|
||||
log::debug!(target: "sync", "Error decoding proof: {:?}", e);
|
||||
debug!(target: "sync", "Error decoding proof: {:?}", e);
|
||||
return ImportResult::BadResponse
|
||||
},
|
||||
};
|
||||
let (values, complete) =
|
||||
match self.client.verify_range_proof(self.target_root, proof, &self.last_key) {
|
||||
Err(e) => {
|
||||
log::debug!(
|
||||
target: "sync",
|
||||
"StateResponse failed proof verification: {:?}",
|
||||
e,
|
||||
);
|
||||
debug!(target: "sync", "StateResponse failed proof verification: {:?}", e);
|
||||
return ImportResult::BadResponse
|
||||
},
|
||||
Ok(values) => values,
|
||||
};
|
||||
log::debug!(target: "sync", "Imported with {} keys", values.len());
|
||||
debug!(target: "sync", "Imported with {} keys", values.len());
|
||||
|
||||
if let Some(last) = values.last().map(|(k, _)| k) {
|
||||
self.last_key = last.clone();
|
||||
@@ -123,7 +110,7 @@ impl<B: BlockT> StateSync<B> {
|
||||
self.imported_bytes += proof_size;
|
||||
complete
|
||||
} else {
|
||||
log::debug!(
|
||||
debug!(
|
||||
target: "sync",
|
||||
"Importing state from {:?} to {:?}",
|
||||
response.entries.last().map(|e| sp_core::hexdisplay::HexDisplay::from(&e.key)),
|
||||
@@ -142,12 +129,9 @@ impl<B: BlockT> StateSync<B> {
|
||||
if complete {
|
||||
self.complete = true;
|
||||
ImportResult::Import(
|
||||
self.target_block.clone(),
|
||||
self.target_block,
|
||||
self.target_header.clone(),
|
||||
ImportedState {
|
||||
block: self.target_block.clone(),
|
||||
state: std::mem::take(&mut self.state),
|
||||
},
|
||||
ImportedState { block: self.target_block, state: std::mem::take(&mut self.state) },
|
||||
)
|
||||
} else {
|
||||
ImportResult::Continue(self.next_request())
|
||||
@@ -170,12 +154,12 @@ impl<B: BlockT> StateSync<B> {
|
||||
|
||||
/// Returns target block number.
|
||||
pub fn target_block_num(&self) -> NumberFor<B> {
|
||||
self.target_header.number().clone()
|
||||
*self.target_header.number()
|
||||
}
|
||||
|
||||
/// Returns target block hash.
|
||||
pub fn target(&self) -> B::Hash {
|
||||
self.target_block.clone()
|
||||
self.target_block
|
||||
}
|
||||
|
||||
/// Returns state sync estimated progress.
|
||||
|
||||
@@ -66,7 +66,7 @@ impl<B: BlockT> WarpSync<B> {
|
||||
authorities: warp_sync_provider.current_authorities(),
|
||||
last_hash,
|
||||
};
|
||||
WarpSync { client, warp_sync_provider, phase, total_proof_bytes: 0 }
|
||||
Self { client, warp_sync_provider, phase, total_proof_bytes: 0 }
|
||||
}
|
||||
|
||||
/// Validate and import a state reponse.
|
||||
@@ -132,8 +132,7 @@ impl<B: BlockT> WarpSync<B> {
|
||||
pub fn next_warp_poof_request(&self) -> Option<WarpProofRequest<B>> {
|
||||
match &self.phase {
|
||||
Phase::State(_) => None,
|
||||
Phase::WarpProof { last_hash, .. } =>
|
||||
Some(WarpProofRequest { begin: last_hash.clone() }),
|
||||
Phase::WarpProof { last_hash, .. } => Some(WarpProofRequest { begin: *last_hash }),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -132,7 +132,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
/// Returns a `NetworkWorker` that implements `Future` and must be regularly polled in order
|
||||
/// for the network processing to advance. From it, you can extract a `NetworkService` using
|
||||
/// `worker.service()`. The `NetworkService` can be shared through the codebase.
|
||||
pub fn new(mut params: Params<B, H>) -> Result<NetworkWorker<B, H>, Error> {
|
||||
pub fn new(mut params: Params<B, H>) -> Result<Self, Error> {
|
||||
// Ensure the listen addresses are consistent with the transport.
|
||||
ensure_addresses_consistent_with_transport(
|
||||
params.network_config.listen_addresses.iter(),
|
||||
@@ -218,9 +218,9 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
|
||||
// Process the bootnodes.
|
||||
for bootnode in params.network_config.boot_nodes.iter() {
|
||||
bootnodes.push(bootnode.peer_id.clone());
|
||||
boot_node_ids.insert(bootnode.peer_id.clone());
|
||||
known_addresses.push((bootnode.peer_id.clone(), bootnode.multiaddr.clone()));
|
||||
bootnodes.push(bootnode.peer_id);
|
||||
boot_node_ids.insert(bootnode.peer_id);
|
||||
known_addresses.push((bootnode.peer_id, bootnode.multiaddr.clone()));
|
||||
}
|
||||
|
||||
let boot_node_ids = Arc::new(boot_node_ids);
|
||||
@@ -230,7 +230,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
if let Some(other) = known_addresses.iter().find(|o| o.1 == *addr && o.0 != *peer_id) {
|
||||
Err(Error::DuplicateBootnode {
|
||||
address: addr.clone(),
|
||||
first_id: peer_id.clone(),
|
||||
first_id: *peer_id,
|
||||
second_id: other.0.clone(),
|
||||
})
|
||||
} else {
|
||||
@@ -364,7 +364,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
}
|
||||
};
|
||||
|
||||
let mut builder = SwarmBuilder::new(transport, behaviour, local_peer_id.clone())
|
||||
let mut builder = SwarmBuilder::new(transport, behaviour, local_peer_id)
|
||||
.connection_limits(
|
||||
ConnectionLimits::default()
|
||||
.with_max_established_per_peer(Some(crate::MAX_CONNECTIONS_PER_PEER as u32))
|
||||
@@ -743,7 +743,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
|
||||
sink.clone()
|
||||
} else {
|
||||
// Notification silently discarded, as documented.
|
||||
log::debug!(
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Attempted to send notification on missing or closed substream: {}, {:?}",
|
||||
target, protocol,
|
||||
@@ -762,9 +762,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"External API => Notification({:?}, {:?}, {} bytes)",
|
||||
target,
|
||||
protocol,
|
||||
message.len()
|
||||
target, protocol, message.len()
|
||||
);
|
||||
trace!(target: "sub-libp2p", "Handler({:?}) <= Sync notification", target);
|
||||
sink.send_sync_notification(message);
|
||||
@@ -1260,7 +1258,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
|
||||
|
||||
impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle for NetworkService<B, H> {
|
||||
fn is_major_syncing(&mut self) -> bool {
|
||||
NetworkService::is_major_syncing(self)
|
||||
Self::is_major_syncing(self)
|
||||
}
|
||||
|
||||
fn is_offline(&mut self) -> bool {
|
||||
@@ -1280,11 +1278,11 @@ impl<'a, B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle for &'a Netwo
|
||||
|
||||
impl<B: BlockT, H: ExHashT> sc_consensus::JustificationSyncLink<B> for NetworkService<B, H> {
|
||||
fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
|
||||
NetworkService::request_justification(self, hash, number);
|
||||
Self::request_justification(self, hash, number);
|
||||
}
|
||||
|
||||
fn clear_justification_requests(&self) {
|
||||
NetworkService::clear_justification_requests(self);
|
||||
Self::clear_justification_requests(self);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1300,7 +1298,7 @@ where
|
||||
|
||||
/// Returns the local Peer ID.
|
||||
fn local_peer_id(&self) -> PeerId {
|
||||
self.local_peer_id.clone()
|
||||
self.local_peer_id
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1363,9 +1361,7 @@ impl<'a> NotificationSenderReady<'a> {
|
||||
trace!(
|
||||
target: "sub-libp2p",
|
||||
"External API => Notification({:?}, {}, {} bytes)",
|
||||
self.peer_id,
|
||||
self.protocol_name,
|
||||
notification.len()
|
||||
self.peer_id, self.protocol_name, notification.len(),
|
||||
);
|
||||
trace!(target: "sub-libp2p", "Handler({:?}) <= Async notification", self.peer_id);
|
||||
|
||||
@@ -1477,9 +1473,7 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
match result {
|
||||
Ok(()) => {},
|
||||
Err(light_client_requests::sender::SendRequestError::TooManyRequests) => {
|
||||
log::warn!(
|
||||
"Couldn't start light client request: too many pending requests"
|
||||
);
|
||||
warn!("Couldn't start light client request: too many pending requests");
|
||||
},
|
||||
}
|
||||
|
||||
@@ -1771,7 +1765,7 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
if let Some(s) = peers_notifications_sinks.get_mut(&(remote, protocol)) {
|
||||
*s = notifications_sink;
|
||||
} else {
|
||||
log::error!(
|
||||
error!(
|
||||
target: "sub-libp2p",
|
||||
"NotificationStreamReplaced for non-existing substream"
|
||||
);
|
||||
@@ -1931,18 +1925,16 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
},
|
||||
Poll::Ready(SwarmEvent::UnreachableAddr { peer_id, address, error, .. }) => {
|
||||
trace!(
|
||||
target: "sub-libp2p", "Libp2p => Failed to reach {:?} through {:?}: {}",
|
||||
peer_id,
|
||||
address,
|
||||
error,
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => Failed to reach {:?} through {:?}: {}",
|
||||
peer_id, address, error,
|
||||
);
|
||||
|
||||
if this.boot_node_ids.contains(&peer_id) {
|
||||
if let PendingConnectionError::InvalidPeerId = error {
|
||||
error!(
|
||||
"💔 The bootnode you want to connect to at `{}` provided a different peer ID than the one you expect: `{}`.",
|
||||
address,
|
||||
peer_id,
|
||||
address, peer_id,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1980,8 +1972,11 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
send_back_addr,
|
||||
error,
|
||||
}) => {
|
||||
debug!(target: "sub-libp2p", "Libp2p => IncomingConnectionError({},{}): {}",
|
||||
local_addr, send_back_addr, error);
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => IncomingConnectionError({},{}): {}",
|
||||
local_addr, send_back_addr, error,
|
||||
);
|
||||
if let Some(metrics) = this.metrics.as_ref() {
|
||||
let reason = match error {
|
||||
PendingConnectionError::ConnectionLimit(_) => "limit-reached",
|
||||
@@ -1997,8 +1992,11 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
}
|
||||
},
|
||||
Poll::Ready(SwarmEvent::BannedPeer { peer_id, endpoint }) => {
|
||||
debug!(target: "sub-libp2p", "Libp2p => BannedPeer({}). Connected via {:?}.",
|
||||
peer_id, endpoint);
|
||||
debug!(
|
||||
target: "sub-libp2p",
|
||||
"Libp2p => BannedPeer({}). Connected via {:?}.",
|
||||
peer_id, endpoint,
|
||||
);
|
||||
if let Some(metrics) = this.metrics.as_ref() {
|
||||
metrics
|
||||
.incoming_connections_errors_total
|
||||
|
||||
@@ -142,7 +142,7 @@ impl OutChannels {
|
||||
let metrics =
|
||||
if let Some(registry) = registry { Some(Metrics::register(registry)?) } else { None };
|
||||
|
||||
Ok(OutChannels { event_streams: Vec::new(), metrics: Arc::new(metrics) })
|
||||
Ok(Self { event_streams: Vec::new(), metrics: Arc::new(metrics) })
|
||||
}
|
||||
|
||||
/// Adds a new [`Sender`] to the collection.
|
||||
|
||||
@@ -356,15 +356,13 @@ fn lots_of_incoming_peers_works() {
|
||||
..config::NetworkConfiguration::new_local()
|
||||
});
|
||||
|
||||
let main_node_peer_id = main_node.local_peer_id().clone();
|
||||
let main_node_peer_id = *main_node.local_peer_id();
|
||||
|
||||
// We spawn background tasks and push them in this `Vec`. They will all be waited upon before
|
||||
// this test ends.
|
||||
let mut background_tasks_to_wait = Vec::new();
|
||||
|
||||
for _ in 0..32 {
|
||||
let main_node_peer_id = main_node_peer_id.clone();
|
||||
|
||||
let (_dialing_node, event_stream) = build_test_full_node(config::NetworkConfiguration {
|
||||
listen_addresses: vec![],
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
@@ -374,7 +372,7 @@ fn lots_of_incoming_peers_works() {
|
||||
set_config: config::SetConfig {
|
||||
reserved_nodes: vec![config::MultiaddrWithPeerId {
|
||||
multiaddr: listen_addr.clone(),
|
||||
peer_id: main_node_peer_id.clone(),
|
||||
peer_id: main_node_peer_id,
|
||||
}],
|
||||
..Default::default()
|
||||
},
|
||||
|
||||
@@ -29,7 +29,7 @@ use futures::{
|
||||
channel::{mpsc, oneshot},
|
||||
stream::StreamExt,
|
||||
};
|
||||
use log::debug;
|
||||
use log::{debug, trace};
|
||||
use lru::LruCache;
|
||||
use prost::Message;
|
||||
use sp_runtime::{generic::BlockId, traits::Block as BlockT};
|
||||
@@ -166,7 +166,7 @@ impl<B: BlockT> StateRequestHandler<B> {
|
||||
},
|
||||
}
|
||||
|
||||
log::trace!(
|
||||
trace!(
|
||||
target: LOG_TARGET,
|
||||
"Handling state request from {}: Block {:?}, Starting at {:?}, no_proof={}",
|
||||
peer,
|
||||
@@ -201,7 +201,7 @@ impl<B: BlockT> StateRequestHandler<B> {
|
||||
}
|
||||
}
|
||||
|
||||
log::trace!(
|
||||
trace!(
|
||||
target: LOG_TARGET,
|
||||
"StateResponse contains {} keys, {}, proof nodes, complete={}, from {:?} to {:?}",
|
||||
response.entries.len(),
|
||||
|
||||
@@ -46,14 +46,14 @@ fn prepare_good_block() -> (TestClient, Hash, u64, PeerId, IncomingBlock<Block>)
|
||||
client,
|
||||
hash,
|
||||
number,
|
||||
peer_id.clone(),
|
||||
peer_id,
|
||||
IncomingBlock {
|
||||
hash,
|
||||
header,
|
||||
body: Some(Vec::new()),
|
||||
indexed_body: None,
|
||||
justifications,
|
||||
origin: Some(peer_id.clone()),
|
||||
origin: Some(peer_id),
|
||||
allow_missing_state: false,
|
||||
import_existing: false,
|
||||
state: None,
|
||||
|
||||
@@ -426,7 +426,7 @@ impl Peerset {
|
||||
// We want reputations to be up-to-date before adjusting them.
|
||||
self.update_time();
|
||||
|
||||
let mut reputation = self.data.peer_reputation(peer_id.clone());
|
||||
let mut reputation = self.data.peer_reputation(peer_id);
|
||||
reputation.add_reputation(change.value);
|
||||
if reputation.reputation() >= BANNED_THRESHOLD {
|
||||
trace!(target: "peerset", "Report {}: {:+} to {}. Reason: {}",
|
||||
@@ -486,7 +486,7 @@ impl Peerset {
|
||||
reput.saturating_sub(diff)
|
||||
}
|
||||
|
||||
let mut peer_reputation = self.data.peer_reputation(peer_id.clone());
|
||||
let mut peer_reputation = self.data.peer_reputation(peer_id);
|
||||
|
||||
let before = peer_reputation.reputation();
|
||||
let after = reput_tick(before);
|
||||
@@ -920,7 +920,7 @@ mod tests {
|
||||
assert_eq!(Stream::poll_next(Pin::new(&mut peerset), cx), Poll::Pending);
|
||||
|
||||
// Check that an incoming connection from that node gets refused.
|
||||
peerset.incoming(SetId::from(0), peer_id.clone(), IncomingIndex(1));
|
||||
peerset.incoming(SetId::from(0), peer_id, IncomingIndex(1));
|
||||
if let Poll::Ready(msg) = Stream::poll_next(Pin::new(&mut peerset), cx) {
|
||||
assert_eq!(msg.unwrap(), Message::Reject(IncomingIndex(1)));
|
||||
} else {
|
||||
@@ -931,7 +931,7 @@ mod tests {
|
||||
thread::sleep(Duration::from_millis(1500));
|
||||
|
||||
// Try again. This time the node should be accepted.
|
||||
peerset.incoming(SetId::from(0), peer_id.clone(), IncomingIndex(2));
|
||||
peerset.incoming(SetId::from(0), peer_id, IncomingIndex(2));
|
||||
while let Poll::Ready(msg) = Stream::poll_next(Pin::new(&mut peerset), cx) {
|
||||
assert_eq!(msg.unwrap(), Message::Accept(IncomingIndex(2)));
|
||||
}
|
||||
@@ -965,7 +965,7 @@ mod tests {
|
||||
// Check that an incoming connection from that node gets refused.
|
||||
// This is already tested in other tests, but it is done again here because it doesn't
|
||||
// hurt.
|
||||
peerset.incoming(SetId::from(0), peer_id.clone(), IncomingIndex(1));
|
||||
peerset.incoming(SetId::from(0), peer_id, IncomingIndex(1));
|
||||
if let Poll::Ready(msg) = Stream::poll_next(Pin::new(&mut peerset), cx) {
|
||||
assert_eq!(msg.unwrap(), Message::Reject(IncomingIndex(1)));
|
||||
} else {
|
||||
|
||||
@@ -167,7 +167,7 @@ impl PeersState {
|
||||
/// Returns an object that grants access to the reputation value of a peer.
|
||||
pub fn peer_reputation(&mut self, peer_id: PeerId) -> Reputation {
|
||||
if !self.nodes.contains_key(&peer_id) {
|
||||
self.nodes.insert(peer_id.clone(), Node::new(self.sets.len()));
|
||||
self.nodes.insert(peer_id, Node::new(self.sets.len()));
|
||||
}
|
||||
|
||||
let entry = match self.nodes.entry(peer_id) {
|
||||
@@ -256,7 +256,7 @@ impl PeersState {
|
||||
}
|
||||
Some(to_try)
|
||||
})
|
||||
.map(|(peer_id, _)| peer_id.clone());
|
||||
.map(|(peer_id, _)| *peer_id);
|
||||
|
||||
outcome.map(move |peer_id| NotConnectedPeer {
|
||||
state: self,
|
||||
@@ -275,7 +275,7 @@ impl PeersState {
|
||||
/// Has no effect if the node was already in the group.
|
||||
pub fn add_no_slot_node(&mut self, set: usize, peer_id: PeerId) {
|
||||
// Reminder: `HashSet::insert` returns false if the node was already in the set
|
||||
if !self.sets[set].no_slot_nodes.insert(peer_id.clone()) {
|
||||
if !self.sets[set].no_slot_nodes.insert(peer_id) {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -223,8 +223,7 @@ fn system_local_listen_addresses_works() {
|
||||
#[test]
|
||||
fn system_peers() {
|
||||
let peer_id = PeerId::random();
|
||||
let req = api(Status { peer_id: peer_id.clone(), peers: 1, is_syncing: false, is_dev: true })
|
||||
.system_peers();
|
||||
let req = api(Status { peer_id, peers: 1, is_syncing: false, is_dev: true }).system_peers();
|
||||
let res = executor::block_on(req).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
|
||||
Reference in New Issue
Block a user