pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-13 09:21:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Run cargo fmt on the whole code base (#9394)

Browse Source 
* Run cargo fmt on the whole code base

* Second run

* Add CI check

* Fix compilation

* More unnecessary braces

* Handle weights

* Use --all

* Use correct attributes...

* Fix UI tests

* AHHHHHHHHH

* 🤦

* Docs

* Fix compilation

* 🤷

* Please stop

* 🤦 x 2

* More

* make rustfmt.toml consistent with polkadot

Co-authored-by: André Silva <andrerfosilva@gmail.com>
This commit is contained in:
Bastian Köcher
2021-07-21 16:32:32 +02:00
committed by GitHub
parent d451c38c1c
commit 7b56ab15b4
1010 changed files with 53339 additions and 51208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/authority-discovery/src/interval.rs
+7 -11
View File
@@ -16,13 +16,13 @@
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
use futures::stream::Stream;
use futures::future::FutureExt;
use futures::ready;
use futures::{future::FutureExt, ready, stream::Stream};
use futures_timer::Delay;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use std::{
pin::Pin,
task::{Context, Poll},
time::Duration,
};
/// Exponentially increasing interval
///
@@ -37,11 +37,7 @@ impl ExpIncInterval {
/// Create a new [`ExpIncInterval`].
pub fn new(start: Duration, max: Duration) -> Self {
let delay = Delay::new(start);
Self {
max,
next: start * 2,
delay,
}
Self { max, next: start * 2, delay }
}
/// Fast forward the exponentially increasing interval to the configured maximum.
substrate/client/authority-discovery/src/lib.rs
+12 -14
View File
@@ -26,18 +26,23 @@
//!
//! See [`Worker`] and [`Service`] for more documentation.
pub use crate::{service::Service, worker::{NetworkProvider, Worker, Role}};
pub use crate::{
service::Service,
worker::{NetworkProvider, Role, Worker},
};
use std::{sync::Arc, time::Duration};
use futures::channel::{mpsc, oneshot};
use futures::Stream;
use futures::{
channel::{mpsc, oneshot},
Stream,
};
use sc_client_api::blockchain::HeaderBackend;
use sc_network::{DhtEvent, Multiaddr, PeerId};
use sp_api::ProvideRuntimeApi;
use sp_authority_discovery::{AuthorityDiscoveryApi, AuthorityId};
use sp_runtime::traits::Block as BlockT;
use sp_api::ProvideRuntimeApi;
mod error;
mod interval;
@@ -141,15 +146,8 @@ where
{
let (to_worker, from_service) = mpsc::channel(0);
let worker = Worker::new(
from_service,
client,
network,
dht_event_rx,
role,
prometheus_registry,
config,
);
let worker =
Worker::new(from_service, client, network, dht_event_rx, role, prometheus_registry, config);
let service = Service::new(to_worker);
(worker, service)
@@ -160,5 +158,5 @@ pub(crate) enum ServicetoWorkerMsg {
/// See [`Service::get_addresses_by_authority_id`].
GetAddressesByAuthorityId(AuthorityId, oneshot::Sender<Option<Vec<Multiaddr>>>),
/// See [`Service::get_authority_id_by_peer_id`].
GetAuthorityIdByPeerId(PeerId, oneshot::Sender<Option<AuthorityId>>)
GetAuthorityIdByPeerId(PeerId, oneshot::Sender<Option<AuthorityId>>),
}
substrate/client/authority-discovery/src/service.rs
+9 -6
View File
@@ -20,8 +20,10 @@ use std::fmt::Debug;
use crate::ServicetoWorkerMsg;
use futures::channel::{mpsc, oneshot};
use futures::SinkExt;
use futures::{
channel::{mpsc, oneshot},
SinkExt,
};
use sc_network::{Multiaddr, PeerId};
use sp_authority_discovery::AuthorityId;
@@ -42,9 +44,7 @@ impl Debug for Service {
/// [`crate::Worker`]'s local address cache for a given [`AuthorityId`].
impl Service {
pub(crate) fn new(to_worker: mpsc::Sender<ServicetoWorkerMsg>) -> Self {
Self {
to_worker,
}
Self { to_worker }
}
/// Get the addresses for the given [`AuthorityId`] from the local address
@@ -59,7 +59,10 @@ impl Service {
/// enforced today, given that there are still authorities out there
/// publishing the addresses of their sentry nodes on the DHT. In the future
/// this guarantee can be provided.
pub async fn get_addresses_by_authority_id(&mut self, authority: AuthorityId) -> Option<Vec<Multiaddr>> {
pub async fn get_addresses_by_authority_id(
&mut self,
authority: AuthorityId,
) -> Option<Vec<Multiaddr>> {
let (tx, rx) = oneshot::channel();
self.to_worker
substrate/client/authority-discovery/src/tests.rs
+16 -8
View File
@@ -16,15 +16,24 @@
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
use crate::{new_worker_and_service, worker::{tests::{TestApi, TestNetwork}, Role}};
use crate::{
new_worker_and_service,
worker::{
tests::{TestApi, TestNetwork},
Role,
},
};
use std::sync::Arc;
use futures::{channel::mpsc::channel, executor::LocalPool, task::LocalSpawn};
use libp2p::core::{multiaddr::{Multiaddr, Protocol}, PeerId};
use libp2p::core::{
multiaddr::{Multiaddr, Protocol},
PeerId,
};
use std::sync::Arc;
use sp_authority_discovery::AuthorityId;
use sp_core::crypto::key_types;
use sp_keystore::{CryptoStore, testing::KeyStore};
use sp_keystore::{testing::KeyStore, CryptoStore};
#[test]
fn get_addresses_and_authority_id() {
@@ -44,13 +53,12 @@ fn get_addresses_and_authority_id() {
});
let remote_peer_id = PeerId::random();
let remote_addr = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333".parse::<Multiaddr>()
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()));
let test_api = Arc::new(TestApi {
authorities: vec![],
});
let test_api = Arc::new(TestApi { authorities: vec![] });
let (mut worker, mut service) = new_worker_and_service(
test_api,
substrate/client/authority-discovery/src/worker.rs
+125 -145
View File
@@ -16,43 +16,49 @@
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
use crate::{error::{Error, Result}, interval::ExpIncInterval, ServicetoWorkerMsg};
use crate::{
error::{Error, Result},
interval::ExpIncInterval,
ServicetoWorkerMsg,
};
use std::collections::{HashMap, HashSet};
use std::convert::TryInto;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use std::{
collections::{HashMap, HashSet},
convert::TryInto,
marker::PhantomData,
sync::Arc,
time::Duration,
};
use futures::channel::mpsc;
use futures::{future, FutureExt, Stream, StreamExt, stream::Fuse};
use futures::{channel::mpsc, future, stream::Fuse, FutureExt, Stream, StreamExt};
use addr_cache::AddrCache;
use async_trait::async_trait;
use codec::Decode;
use ip_network::IpNetwork;
use libp2p::{core::multiaddr, multihash::{Multihash, Hasher}};
use libp2p::{
core::multiaddr,
multihash::{Hasher, Multihash},
};
use log::{debug, error, log_enabled};
use prometheus_endpoint::{Counter, CounterVec, Gauge, Opts, U64, register};
use prometheus_endpoint::{register, Counter, CounterVec, Gauge, Opts, U64};
use prost::Message;
use rand::{seq::SliceRandom, thread_rng};
use sc_client_api::blockchain::HeaderBackend;
use sc_network::{
DhtEvent,
ExHashT,
Multiaddr,
NetworkStateInfo,
PeerId,
use sc_network::{DhtEvent, ExHashT, Multiaddr, NetworkStateInfo, PeerId};
use sp_api::ProvideRuntimeApi;
use sp_authority_discovery::{
AuthorityDiscoveryApi, AuthorityId, AuthorityPair, AuthoritySignature,
};
use sp_authority_discovery::{AuthorityDiscoveryApi, AuthorityId, AuthoritySignature, AuthorityPair};
use sp_core::crypto::{key_types, CryptoTypePublicPair, Pair};
use sp_keystore::CryptoStore;
use sp_runtime::{traits::Block as BlockT, generic::BlockId};
use sp_api::ProvideRuntimeApi;
use sp_runtime::{generic::BlockId, traits::Block as BlockT};
mod addr_cache;
/// Dht payload schemas generated from Protobuf definitions via Prost crate in build.rs.
mod schema { include!(concat!(env!("OUT_DIR"), "/authority_discovery.rs")); }
mod schema {
include!(concat!(env!("OUT_DIR"), "/authority_discovery.rs"));
}
#[cfg(test)]
pub mod tests;
@@ -72,7 +78,6 @@ pub enum Role {
Discover,
}
/// An authority discovery [`Worker`] can publish the local node's addresses as well as discover
/// those of other nodes via a Kademlia DHT.
///
@@ -141,8 +146,7 @@ where
Block: BlockT + Unpin + 'static,
Network: NetworkProvider,
Client: ProvideRuntimeApi<Block> + Send + Sync + 'static + HeaderBackend<Block>,
<Client as ProvideRuntimeApi<Block>>::Api:
AuthorityDiscoveryApi<Block>,
<Client as ProvideRuntimeApi<Block>>::Api: AuthorityDiscoveryApi<Block>,
DhtEventStream: Stream<Item = DhtEvent> + Unpin,
{
/// Construct a [`Worker`].
@@ -161,33 +165,24 @@ where
// thus timely retries are not needed. For this reasoning use an exponentially increasing
// interval for `publish_interval`, `query_interval` and `priority_group_set_interval`
// instead of a constant interval.
let publish_interval = ExpIncInterval::new(
Duration::from_secs(2),
config.max_publish_interval,
);
let query_interval = ExpIncInterval::new(
Duration::from_secs(2),
config.max_query_interval,
);
let publish_interval =
ExpIncInterval::new(Duration::from_secs(2), config.max_publish_interval);
let query_interval = ExpIncInterval::new(Duration::from_secs(2), config.max_query_interval);
// An `ExpIncInterval` is overkill here because the interval is constant, but consistency
// is more simple.
let publish_if_changed_interval = ExpIncInterval::new(
config.keystore_refresh_interval,
config.keystore_refresh_interval
);
let publish_if_changed_interval =
ExpIncInterval::new(config.keystore_refresh_interval, config.keystore_refresh_interval);
let addr_cache = AddrCache::new();
let metrics = match prometheus_registry {
Some(registry) => {
match Metrics::register(&registry) {
Ok(metrics) => Some(metrics),
Err(e) => {
error!(target: LOG_TARGET, "Failed to register metrics: {:?}", e);
None
},
}
Some(registry) => match Metrics::register(&registry) {
Ok(metrics) => Some(metrics),
Err(e) => {
error!(target: LOG_TARGET, "Failed to register metrics: {:?}", e);
None
},
},
None => None,
};
@@ -262,23 +257,23 @@ where
let _ = sender.send(
self.addr_cache.get_addresses_by_authority_id(&authority).map(Clone::clone),
);
}
},
ServicetoWorkerMsg::GetAuthorityIdByPeerId(peer_id, sender) => {
let _ = sender.send(
self.addr_cache.get_authority_id_by_peer_id(&peer_id).map(Clone::clone),
);
}
let _ = sender
.send(self.addr_cache.get_authority_id_by_peer_id(&peer_id).map(Clone::clone));
},
}
}
fn addresses_to_publish(&self) -> impl Iterator<Item = Multiaddr> {
let peer_id: Multihash = self.network.local_peer_id().into();
let publish_non_global_ips = self.publish_non_global_ips;
self.network.external_addresses()
self.network
.external_addresses()
.into_iter()
.filter(move |a| {
if publish_non_global_ips {
return true;
return true
}
a.iter().all(|p| match p {
@@ -321,9 +316,9 @@ where
if let Some(metrics) = &self.metrics {
metrics.publish.inc();
metrics.amount_addresses_last_published.set(
addresses.len().try_into().unwrap_or(std::u64::MAX),
);
metrics
.amount_addresses_last_published
.set(addresses.len().try_into().unwrap_or(std::u64::MAX));
}
let mut serialized_addresses = vec![];
@@ -332,30 +327,26 @@ where
.map_err(Error::EncodingProto)?;
let keys_vec = keys.iter().cloned().collect::<Vec<_>>();
let signatures = key_store.sign_with_all(
key_types::AUTHORITY_DISCOVERY,
keys_vec.clone(),
serialized_addresses.as_slice(),
).await.map_err(|_| Error::Signing)?;
let signatures = key_store
.sign_with_all(
key_types::AUTHORITY_DISCOVERY,
keys_vec.clone(),
serialized_addresses.as_slice(),
)
.await
.map_err(|_| Error::Signing)?;
for (sign_result, key) in signatures.into_iter().zip(keys_vec.iter()) {
let mut signed_addresses = vec![];
// Verify that all signatures exist for all provided keys.
let signature = sign_result.ok()
.flatten()
.ok_or_else(|| Error::MissingSignature(key.clone()))?;
schema::SignedAuthorityAddresses {
addresses: serialized_addresses.clone(),
signature,
}
.encode(&mut signed_addresses)
let signature =
sign_result.ok().flatten().ok_or_else(|| Error::MissingSignature(key.clone()))?;
schema::SignedAuthorityAddresses { addresses: serialized_addresses.clone(), signature }
.encode(&mut signed_addresses)
.map_err(Error::EncodingProto)?;
self.network.put_value(
hash_authority_id(key.1.as_ref()),
signed_addresses,
);
self.network.put_value(hash_authority_id(key.1.as_ref()), signed_addresses);
}
self.latest_published_keys = keys;
@@ -367,11 +358,11 @@ where
let id = BlockId::hash(self.client.info().best_hash);
let local_keys = match &self.role {
Role::PublishAndDiscover(key_store) => {
key_store.sr25519_public_keys(
key_types::AUTHORITY_DISCOVERY
).await.into_iter().collect::<HashSet<_>>()
},
Role::PublishAndDiscover(key_store) => key_store
.sr25519_public_keys(key_types::AUTHORITY_DISCOVERY)
.await
.into_iter()
.collect::<HashSet<_>>(),
Role::Discover => HashSet::new(),
};
@@ -393,9 +384,9 @@ where
self.in_flight_lookups.clear();
if let Some(metrics) = &self.metrics {
metrics.requests_pending.set(
self.pending_lookups.len().try_into().unwrap_or(std::u64::MAX),
);
metrics
.requests_pending
.set(self.pending_lookups.len().try_into().unwrap_or(std::u64::MAX));
}
Ok(())
@@ -408,15 +399,14 @@ where
None => return,
};
let hash = hash_authority_id(authority_id.as_ref());
self.network
.get_value(&hash);
self.network.get_value(&hash);
self.in_flight_lookups.insert(hash, authority_id);
if let Some(metrics) = &self.metrics {
metrics.requests.inc();
metrics.requests_pending.set(
self.pending_lookups.len().try_into().unwrap_or(std::u64::MAX),
);
metrics
.requests_pending
.set(self.pending_lookups.len().try_into().unwrap_or(std::u64::MAX));
}
}
}
@@ -431,10 +421,7 @@ where
if log_enabled!(log::Level::Debug) {
let hashes: Vec<_> = v.iter().map(|(hash, _value)| hash.clone()).collect();
debug!(
target: LOG_TARGET,
"Value for hash '{:?}' found on Dht.", hashes,
);
debug!(target: LOG_TARGET, "Value for hash '{:?}' found on Dht.", hashes,);
}
if let Err(e) = self.handle_dht_value_found_event(v) {
@@ -442,22 +429,16 @@ where
metrics.handle_value_found_event_failure.inc();
}
debug!(
target: LOG_TARGET,
"Failed to handle Dht value found event: {:?}", e,
);
debug!(target: LOG_TARGET, "Failed to handle Dht value found event: {:?}", e,);
}
}
},
DhtEvent::ValueNotFound(hash) => {
if let Some(metrics) = &self.metrics {
metrics.dht_event_received.with_label_values(&["value_not_found"]).inc();
}
if self.in_flight_lookups.remove(&hash).is_some() {
debug!(
target: LOG_TARGET,
"Value for hash '{:?}' not found on Dht.", hash
)
debug!(target: LOG_TARGET, "Value for hash '{:?}' not found on Dht.", hash)
} else {
debug!(
target: LOG_TARGET,
@@ -475,21 +456,15 @@ where
metrics.dht_event_received.with_label_values(&["value_put"]).inc();
}
debug!(
target: LOG_TARGET,
"Successfully put hash '{:?}' on Dht.", hash,
)
debug!(target: LOG_TARGET, "Successfully put hash '{:?}' on Dht.", hash,)
},
DhtEvent::ValuePutFailed(hash) => {
if let Some(metrics) = &self.metrics {
metrics.dht_event_received.with_label_values(&["value_put_failed"]).inc();
}
debug!(
target: LOG_TARGET,
"Failed to put hash '{:?}' on Dht.", hash
)
}
debug!(target: LOG_TARGET, "Failed to put hash '{:?}' on Dht.", hash)
},
}
}
@@ -498,34 +473,36 @@ where
values: Vec<(libp2p::kad::record::Key, Vec<u8>)>,
) -> Result<()> {
// Ensure `values` is not empty and all its keys equal.
let remote_key = values.iter().fold(Ok(None), |acc, (key, _)| {
match acc {
let remote_key = values
.iter()
.fold(Ok(None), |acc, (key, _)| match acc {
Ok(None) => Ok(Some(key.clone())),
Ok(Some(ref prev_key)) if prev_key != key => Err(
Error::ReceivingDhtValueFoundEventWithDifferentKeys
),
Ok(Some(ref prev_key)) if prev_key != key =>
Err(Error::ReceivingDhtValueFoundEventWithDifferentKeys),
x @ Ok(_) => x,
Err(e) => Err(e),
}
})?.ok_or(Error::ReceivingDhtValueFoundEventWithNoRecords)?;
})?
.ok_or(Error::ReceivingDhtValueFoundEventWithNoRecords)?;
let authority_id: AuthorityId = self.in_flight_lookups
let authority_id: AuthorityId = self
.in_flight_lookups
.remove(&remote_key)
.ok_or(Error::ReceivingUnexpectedRecord)?;
let local_peer_id = self.network.local_peer_id();
let remote_addresses: Vec<Multiaddr> = values.into_iter()
let remote_addresses: Vec<Multiaddr> = values
.into_iter()
.map(|(_k, v)| {
let schema::SignedAuthorityAddresses { signature, addresses } =
schema::SignedAuthorityAddresses::decode(v.as_slice())
.map_err(Error::DecodingProto)?;
.map_err(Error::DecodingProto)?;
let signature = AuthoritySignature::decode(&mut &signature[..])
.map_err(Error::EncodingDecodingScale)?;
if !AuthorityPair::verify(&signature, &addresses, &authority_id) {
return Err(Error::VerifyingDhtPayload);
return Err(Error::VerifyingDhtPayload)
}
let addresses = schema::AuthorityAddresses::decode(addresses.as_slice())
@@ -542,40 +519,41 @@ where
.into_iter()
.flatten()
// Ignore [`Multiaddr`]s without [`PeerId`] and own addresses.
.filter(|addr| addr.iter().any(|protocol| {
// Parse to PeerId first as Multihashes of old and new PeerId
// representation don't equal.
//
// See https://github.com/libp2p/rust-libp2p/issues/555 for
// details.
if let multiaddr::Protocol::P2p(hash) = protocol {
let peer_id = match PeerId::from_multihash(hash) {
Ok(peer_id) => peer_id,
Err(_) => return false, // Discard address.
};
.filter(|addr| {
addr.iter().any(|protocol| {
// Parse to PeerId first as Multihashes of old and new PeerId
// representation don't equal.
//
// See https://github.com/libp2p/rust-libp2p/issues/555 for
// details.
if let multiaddr::Protocol::P2p(hash) = protocol {
let peer_id = match PeerId::from_multihash(hash) {
Ok(peer_id) => peer_id,
Err(_) => return false, // Discard address.
};
// Discard if equal to local peer id, keep if it differs.
return !(peer_id == local_peer_id);
}
// Discard if equal to local peer id, keep if it differs.
return !(peer_id == local_peer_id)
}
false // `protocol` is not a [`Protocol::P2p`], let's keep looking.
}))
false // `protocol` is not a [`Protocol::P2p`], let's keep looking.
})
})
.take(MAX_ADDRESSES_PER_AUTHORITY)
.collect();
if !remote_addresses.is_empty() {
self.addr_cache.insert(authority_id, remote_addresses);
if let Some(metrics) = &self.metrics {
metrics.known_authorities_count.set(
self.addr_cache.num_ids().try_into().unwrap_or(std::u64::MAX)
);
metrics
.known_authorities_count
.set(self.addr_cache.num_ids().try_into().unwrap_or(std::u64::MAX));
}
}
Ok(())
}
/// Retrieve our public keys within the current and next authority set.
//
// A node might have multiple authority discovery keys within its keystore, e.g. an old one and
// one for the upcoming session. In addition it could be participating in the current and (/ or)
// next authority set with two keys. The function does not return all of the local authority
@@ -591,14 +569,16 @@ where
.collect::<HashSet<_>>();
let id = BlockId::hash(client.info().best_hash);
let authorities = client.runtime_api()
let authorities = client
.runtime_api()
.authorities(&id)
.map_err(|e| Error::CallingRuntime(e.into()))?
.into_iter()
.map(std::convert::Into::into)
.collect::<HashSet<_>>();
let intersection = local_pub_keys.intersection(&authorities)
let intersection = local_pub_keys
.intersection(&authorities)
.cloned()
.map(std::convert::Into::into)
.collect();
@@ -655,7 +635,7 @@ impl Metrics {
publish: register(
Counter::new(
"authority_discovery_times_published_total",
"Number of times authority discovery has published external addresses."
"Number of times authority discovery has published external addresses.",
)?,
registry,
)?,
@@ -663,7 +643,7 @@ impl Metrics {
Gauge::new(
"authority_discovery_amount_external_addresses_last_published",
"Number of external addresses published when authority discovery last \
published addresses."
published addresses.",
)?,
registry,
)?,
@@ -671,14 +651,14 @@ impl Metrics {
Counter::new(
"authority_discovery_authority_addresses_requested_total",
"Number of times authority discovery has requested external addresses of a \
single authority."
single authority.",
)?,
registry,
)?,
requests_pending: register(
Gauge::new(
"authority_discovery_authority_address_requests_pending",
"Number of pending authority address requests."
"Number of pending authority address requests.",
)?,
registry,
)?,
@@ -686,7 +666,7 @@ impl Metrics {
CounterVec::new(
Opts::new(
"authority_discovery_dht_event_received",
"Number of dht events received by authority discovery."
"Number of dht events received by authority discovery.",
),
&["name"],
)?,
@@ -695,14 +675,14 @@ impl Metrics {
handle_value_found_event_failure: register(
Counter::new(
"authority_discovery_handle_value_found_event_failure",
"Number of times handling a dht value found event failed."
"Number of times handling a dht value found event failed.",
)?,
registry,
)?,
known_authorities_count: register(
Gauge::new(
"authority_discovery_known_authorities_count",
"Number of authorities known by authority discovery."
"Number of authorities known by authority discovery.",
)?,
registry,
)?,
substrate/client/authority-discovery/src/worker/addr_cache.rs
+51 -26
View File
@@ -19,8 +19,8 @@
use libp2p::core::multiaddr::{Multiaddr, Protocol};
use std::collections::HashMap;
use sp_authority_discovery::AuthorityId;
use sc_network::PeerId;
use sp_authority_discovery::AuthorityId;
/// Cache for [`AuthorityId`] -> [`Vec<Multiaddr>`] and [`PeerId`] -> [`AuthorityId`] mappings.
pub(super) struct AddrCache {
@@ -45,27 +45,34 @@ impl AddrCache {
addresses.sort_unstable_by(|a, b| a.as_ref().cmp(b.as_ref()));
// Insert into `self.peer_id_to_authority_id`.
let peer_ids = addresses.iter()
let peer_ids = addresses
.iter()
.map(|a| peer_id_from_multiaddr(a))
.filter_map(|peer_id| peer_id);
for peer_id in peer_ids.clone() {
let former_auth = match self.peer_id_to_authority_id.insert(peer_id, authority_id.clone()) {
Some(a) if a != authority_id => a,
_ => continue,
};
let former_auth =
match self.peer_id_to_authority_id.insert(peer_id, authority_id.clone()) {
Some(a) if a != authority_id => a,
_ => continue,
};
// PeerId was associated to a different authority id before.
// Remove corresponding authority from `self.authority_id_to_addresses`.
let former_auth_addrs = match self.authority_id_to_addresses.get_mut(&former_auth) {
Some(a) => a,
None => { debug_assert!(false); continue }
None => {
debug_assert!(false);
continue
},
};
former_auth_addrs.retain(|a| peer_id_from_multiaddr(a).map_or(true, |p| p != peer_id));
}
// Insert into `self.authority_id_to_addresses`.
for former_addr in
self.authority_id_to_addresses.insert(authority_id.clone(), addresses.clone()).unwrap_or_default()
for former_addr in self
.authority_id_to_addresses
.insert(authority_id.clone(), addresses.clone())
.unwrap_or_default()
{
// Must remove from `self.peer_id_to_authority_id` any PeerId formerly associated
// to that authority but that can't be found in its new addresses.
@@ -87,7 +94,10 @@ impl AddrCache {
}
/// Returns the addresses for the given [`AuthorityId`].
pub fn get_addresses_by_authority_id(&self, authority_id: &AuthorityId) -> Option<&Vec<Multiaddr>> {
pub fn get_addresses_by_authority_id(
&self,
authority_id: &AuthorityId,
) -> Option<&Vec<Multiaddr>> {
self.authority_id_to_addresses.get(&authority_id)
}
@@ -100,7 +110,9 @@ impl AddrCache {
/// [`AuthorityId`]s.
pub fn retain_ids(&mut self, authority_ids: &Vec<AuthorityId>) {
// The below logic could be replaced by `BtreeMap::drain_filter` once it stabilized.
let authority_ids_to_remove = self.authority_id_to_addresses.iter()
let authority_ids_to_remove = self
.authority_id_to_addresses
.iter()
.filter(|(id, _addresses)| !authority_ids.contains(id))
.map(|entry| entry.0)
.cloned()
@@ -111,7 +123,8 @@ impl AddrCache {
let addresses = self.authority_id_to_addresses.remove(&authority_id_to_remove);
// Remove other entries from `self.peer_id_to_authority_id`.
let peer_ids = addresses.iter()
let peer_ids = addresses
.iter()
.flatten()
.map(|a| peer_id_from_multiaddr(a))
.filter_map(|peer_id| peer_id);
@@ -125,10 +138,12 @@ impl AddrCache {
}
fn peer_id_from_multiaddr(addr: &Multiaddr) -> Option<PeerId> {
addr.iter().last().and_then(|protocol| if let Protocol::P2p(multihash) = protocol {
PeerId::from_multihash(multihash).ok()
} else {
None
addr.iter().last().and_then(|protocol| {
if let Protocol::P2p(multihash) = protocol {
PeerId::from_multihash(multihash).ok()
} else {
None
}
})
}
@@ -159,9 +174,11 @@ mod tests {
fn arbitrary(g: &mut Gen) -> Self {
let seed = (0..32).map(|_| u8::arbitrary(g)).collect::<Vec<_>>();
let peer_id = PeerId::from_multihash(
Multihash::wrap(multihash::Code::Sha2_256.into(), &seed).unwrap()
).unwrap();
let multiaddr = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333".parse::<Multiaddr>()
Multihash::wrap(multihash::Code::Sha2_256.into(), &seed).unwrap(),
)
.unwrap();
let multiaddr = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333"
.parse::<Multiaddr>()
.unwrap()
.with(Protocol::P2p(peer_id.into()));
@@ -176,12 +193,15 @@ mod tests {
fn arbitrary(g: &mut Gen) -> Self {
let seed = (0..32).map(|_| u8::arbitrary(g)).collect::<Vec<_>>();
let peer_id = PeerId::from_multihash(
Multihash::wrap(multihash::Code::Sha2_256.into(), &seed).unwrap()
).unwrap();
let multiaddr1 = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333".parse::<Multiaddr>()
Multihash::wrap(multihash::Code::Sha2_256.into(), &seed).unwrap(),
)
.unwrap();
let multiaddr1 = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333"
.parse::<Multiaddr>()
.unwrap()
.with(Protocol::P2p(peer_id.clone().into()));
let multiaddr2 = "/ip6/2002:db8:0:0:0:0:0:2/tcp/30133".parse::<Multiaddr>()
let multiaddr2 = "/ip6/2002:db8:0:0:0:0:0:2/tcp/30133"
.parse::<Multiaddr>()
.unwrap()
.with(Protocol::P2p(peer_id.into()));
TestMultiaddrsSamePeerCombo(multiaddr1, multiaddr2)
@@ -219,11 +239,13 @@ mod tests {
cache.retain_ids(&vec![first.0, second.0]);
assert_eq!(
None, cache.get_addresses_by_authority_id(&third.0),
None,
cache.get_addresses_by_authority_id(&third.0),
"Expect `get_addresses_by_authority_id` to not return `None` for third authority."
);
assert_eq!(
None, cache.get_authority_id_by_peer_id(&peer_id_from_multiaddr(&third.1).unwrap()),
None,
cache.get_authority_id_by_peer_id(&peer_id_from_multiaddr(&third.1).unwrap()),
"Expect `get_authority_id_by_peer_id` to return `None` for third authority."
);
@@ -253,7 +275,10 @@ mod tests {
let mut cache = AddrCache::new();
cache.insert(authority1.clone(), vec![multiaddr1.clone()]);
cache.insert(authority1.clone(), vec![multiaddr2.clone(), multiaddr3.clone(), multiaddr4.clone()]);
cache.insert(
authority1.clone(),
vec![multiaddr2.clone(), multiaddr3.clone(), multiaddr4.clone()],
);
assert_eq!(
None,
substrate/client/authority-discovery/src/worker/tests.rs
+208 -203
View File
@@ -18,21 +18,26 @@
use crate::worker::schema;
use std::{sync::{Arc, Mutex}, task::Poll};
use std::{
sync::{Arc, Mutex},
task::Poll,
};
use async_trait::async_trait;
use futures::channel::mpsc::{self, channel};
use futures::executor::{block_on, LocalPool};
use futures::future::FutureExt;
use futures::sink::SinkExt;
use futures::task::LocalSpawn;
use libp2p::{kad, core::multiaddr, PeerId};
use futures::{
channel::mpsc::{self, channel},
executor::{block_on, LocalPool},
future::FutureExt,
sink::SinkExt,
task::LocalSpawn,
};
use libp2p::{core::multiaddr, kad, PeerId};
use prometheus_endpoint::prometheus::default_registry;
use sp_api::{ProvideRuntimeApi, ApiRef};
use sp_api::{ApiRef, ProvideRuntimeApi};
use sp_core::crypto::Public;
use sp_keystore::{testing::KeyStore, CryptoStore};
use sp_runtime::traits::{Zero, Block as BlockT, NumberFor};
use sp_runtime::traits::{Block as BlockT, NumberFor, Zero};
use substrate_test_runtime_client::runtime::Block;
use super::*;
@@ -46,9 +51,7 @@ impl ProvideRuntimeApi<Block> for TestApi {
type Api = RuntimeApi;
fn runtime_api<'a>(&'a self) -> ApiRef<'a, Self::Api> {
RuntimeApi {
authorities: self.authorities.clone(),
}.into()
RuntimeApi { authorities: self.authorities.clone() }.into()
}
}
@@ -135,10 +138,7 @@ impl Default for TestNetwork {
let (tx, rx) = mpsc::unbounded();
TestNetwork {
peer_id: PeerId::random(),
external_addresses: vec![
"/ip6/2001:db8::/tcp/30333"
.parse().unwrap(),
],
external_addresses: vec!["/ip6/2001:db8::/tcp/30333".parse().unwrap()],
put_value_call: Default::default(),
get_value_call: Default::default(),
event_sender: tx,
@@ -151,11 +151,17 @@ impl Default for TestNetwork {
impl NetworkProvider for TestNetwork {
fn put_value(&self, key: kad::record::Key, value: Vec<u8>) {
self.put_value_call.lock().unwrap().push((key.clone(), value.clone()));
self.event_sender.clone().unbounded_send(TestNetworkEvent::PutCalled(key, value)).unwrap();
self.event_sender
.clone()
.unbounded_send(TestNetworkEvent::PutCalled(key, value))
.unwrap();
}
fn get_value(&self, key: &kad::record::Key) {
self.get_value_call.lock().unwrap().push(key.clone());
self.event_sender.clone().unbounded_send(TestNetworkEvent::GetCalled(key.clone())).unwrap();
self.event_sender
.clone()
.unbounded_send(TestNetworkEvent::GetCalled(key.clone()))
.unwrap();
}
}
@@ -175,9 +181,8 @@ async fn build_dht_event(
key_store: &KeyStore,
) -> (libp2p::kad::record::Key, Vec<u8>) {
let mut serialized_addresses = vec![];
schema::AuthorityAddresses {
addresses: addresses.into_iter().map(|a| a.to_vec()).collect()
}.encode(&mut serialized_addresses)
schema::AuthorityAddresses { addresses: addresses.into_iter().map(|a| a.to_vec()).collect() }
.encode(&mut serialized_addresses)
.map_err(Error::EncodingProto)
.unwrap();
@@ -192,11 +197,9 @@ async fn build_dht_event(
.unwrap();
let mut signed_addresses = vec![];
schema::SignedAuthorityAddresses {
addresses: serialized_addresses.clone(),
signature,
}
.encode(&mut signed_addresses).unwrap();
schema::SignedAuthorityAddresses { addresses: serialized_addresses.clone(), signature }
.encode(&mut signed_addresses)
.unwrap();
let key = hash_authority_id(&public_key.to_raw_vec());
let value = signed_addresses;
@@ -208,9 +211,7 @@ fn new_registers_metrics() {
let (_dht_event_tx, dht_event_rx) = mpsc::channel(1000);
let network: Arc<TestNetwork> = Arc::new(Default::default());
let key_store = KeyStore::new();
let test_api = Arc::new(TestApi {
authorities: vec![],
});
let test_api = Arc::new(TestApi { authorities: vec![] });
let registry = prometheus_endpoint::Registry::new();
@@ -275,65 +276,67 @@ fn publish_discover_cycle() {
let key_store = KeyStore::new();
let _ = pool.spawner().spawn_local_obj(async move {
let node_a_public = key_store
.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None)
.await
.unwrap();
let test_api = Arc::new(TestApi {
authorities: vec![node_a_public.into()],
});
let _ = pool.spawner().spawn_local_obj(
async move {
let node_a_public = key_store
.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None)
.await
.unwrap();
let test_api = Arc::new(TestApi { authorities: vec![node_a_public.into()] });
let (_to_worker, from_service) = mpsc::channel(0);
let mut worker = Worker::new(
from_service,
test_api,
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(key_store.into()),
None,
Default::default(),
);
let (_to_worker, from_service) = mpsc::channel(0);
let mut worker = Worker::new(
from_service,
test_api,
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(key_store.into()),
None,
Default::default(),
);
worker.publish_ext_addresses(false).await.unwrap();
worker.publish_ext_addresses(false).await.unwrap();
// Expect authority discovery to put a new record onto the dht.
assert_eq!(network.put_value_call.lock().unwrap().len(), 1);
// Expect authority discovery to put a new record onto the dht.
assert_eq!(network.put_value_call.lock().unwrap().len(), 1);
let dht_event = {
let (key, value) = network.put_value_call.lock().unwrap().pop().unwrap();
sc_network::DhtEvent::ValueFound(vec![(key, value)])
};
let dht_event = {
let (key, value) = network.put_value_call.lock().unwrap().pop().unwrap();
sc_network::DhtEvent::ValueFound(vec![(key, value)])
};
// Node B discovering node A's address.
// Node B discovering node A's address.
let (mut dht_event_tx, dht_event_rx) = channel(1000);
let test_api = Arc::new(TestApi {
// Make sure node B identifies node A as an authority.
authorities: vec![node_a_public.into()],
});
let network: Arc<TestNetwork> = Arc::new(Default::default());
let key_store = KeyStore::new();
let (mut dht_event_tx, dht_event_rx) = channel(1000);
let test_api = Arc::new(TestApi {
// Make sure node B identifies node A as an authority.
authorities: vec![node_a_public.into()],
});
let network: Arc<TestNetwork> = Arc::new(Default::default());
let key_store = KeyStore::new();
let (_to_worker, from_service) = mpsc::channel(0);
let mut worker = Worker::new(
from_service,
test_api,
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(key_store.into()),
None,
Default::default(),
);
let (_to_worker, from_service) = mpsc::channel(0);
let mut worker = Worker::new(
from_service,
test_api,
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(key_store.into()),
None,
Default::default(),
);
dht_event_tx.try_send(dht_event.clone()).unwrap();
dht_event_tx.try_send(dht_event.clone()).unwrap();
worker.refill_pending_lookups_queue().await.unwrap();
worker.start_new_lookups();
worker.refill_pending_lookups_queue().await.unwrap();
worker.start_new_lookups();
// Make authority discovery handle the event.
worker.handle_dht_event(dht_event).await;
}.boxed_local().into());
// Make authority discovery handle the event.
worker.handle_dht_event(dht_event).await;
}
.boxed_local()
.into(),
);
pool.run();
}
@@ -345,9 +348,7 @@ fn terminate_when_event_stream_terminates() {
let (dht_event_tx, dht_event_rx) = channel(1000);
let network: Arc<TestNetwork> = Arc::new(Default::default());
let key_store = KeyStore::new();
let test_api = Arc::new(TestApi {
authorities: vec![],
});
let test_api = Arc::new(TestApi { authorities: vec![] });
let (to_worker, from_service) = mpsc::channel(0);
let worker = Worker::new(
@@ -358,7 +359,8 @@ fn terminate_when_event_stream_terminates() {
Role::PublishAndDiscover(key_store.into()),
None,
Default::default(),
).run();
)
.run();
futures::pin_mut!(worker);
block_on(async {
@@ -367,7 +369,8 @@ fn terminate_when_event_stream_terminates() {
// Drop sender side of service channel.
drop(to_worker);
assert_eq!(
Poll::Pending, futures::poll!(&mut worker),
Poll::Pending,
futures::poll!(&mut worker),
"Expect the authority discovery module not to terminate once the \
sender side of the service channel is closed.",
);
@@ -377,7 +380,8 @@ fn terminate_when_event_stream_terminates() {
drop(dht_event_tx);
assert_eq!(
Poll::Ready(()), futures::poll!(&mut worker),
Poll::Ready(()),
futures::poll!(&mut worker),
"Expect the authority discovery module to terminate once the \
sending side of the dht event channel is closed.",
);
@@ -390,14 +394,13 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
let peer_id = PeerId::random();
let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
address.with(multiaddr::Protocol::P2p(
peer_id.into(),
))
address.with(multiaddr::Protocol::P2p(peer_id.into()))
};
let remote_key_store = KeyStore::new();
let remote_public_key: AuthorityId = block_on(
remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None),
).unwrap().into();
let remote_public_key: AuthorityId =
block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap()
.into();
let (mut dht_event_tx, dht_event_rx) = channel(1);
let (network, mut network_events) = {
@@ -407,9 +410,7 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
};
let key_store = KeyStore::new();
let test_api = Arc::new(TestApi {
authorities: vec![remote_public_key.clone()],
});
let test_api = Arc::new(TestApi { authorities: vec![remote_public_key.clone()] });
let mut pool = LocalPool::new();
let (mut to_worker, from_service) = mpsc::channel(1);
@@ -427,30 +428,35 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
//
// As this is a local pool, only one future at a time will have the CPU and
// can make progress until the future returns `Pending`.
let _ = pool.spawner().spawn_local_obj(async move {
// Refilling `pending_lookups` only happens every X minutes. Fast
// forward by calling `refill_pending_lookups_queue` directly.
worker.refill_pending_lookups_queue().await.unwrap();
worker.run().await
}.boxed_local().into());
let _ = pool.spawner().spawn_local_obj(
async move {
// Refilling `pending_lookups` only happens every X minutes. Fast
// forward by calling `refill_pending_lookups_queue` directly.
worker.refill_pending_lookups_queue().await.unwrap();
worker.run().await
}
.boxed_local()
.into(),
);
pool.run_until(async {
// Assert worker to trigger a lookup for the one and only authority.
assert!(matches!(
network_events.next().await,
Some(TestNetworkEvent::GetCalled(_))
));
assert!(matches!(network_events.next().await, Some(TestNetworkEvent::GetCalled(_))));
// Send an event that should generate an error
dht_event_tx.send(DhtEvent::ValueFound(Default::default())).await
dht_event_tx
.send(DhtEvent::ValueFound(Default::default()))
.await
.expect("Channel has capacity of 1.");
// Make previously triggered lookup succeed.
let dht_event = {
let (key, value) = build_dht_event(
vec![remote_multiaddr.clone()],
remote_public_key.clone(), &remote_key_store,
).await;
remote_public_key.clone(),
&remote_key_store,
)
.await;
sc_network::DhtEvent::ValueFound(vec![(key, value)])
};
dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
@@ -458,10 +464,10 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
// Expect authority discovery to function normally, now knowing the
// address for the remote node.
let (sender, addresses) = futures::channel::oneshot::channel();
to_worker.send(ServicetoWorkerMsg::GetAddressesByAuthorityId(
remote_public_key,
sender,
)).await.expect("Channel has capacity of 1.");
to_worker
.send(ServicetoWorkerMsg::GetAddressesByAuthorityId(remote_public_key, sender))
.await
.expect("Channel has capacity of 1.");
assert_eq!(Some(vec![remote_multiaddr]), addresses.await.unwrap());
});
}
@@ -469,23 +475,19 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
#[test]
fn limit_number_of_addresses_added_to_cache_per_authority() {
let remote_key_store = KeyStore::new();
let remote_public = block_on(remote_key_store
.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap();
let remote_public =
block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap();
let addresses = (0..100).map(|_| {
let peer_id = PeerId::random();
let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
address.with(multiaddr::Protocol::P2p(
peer_id.into(),
))
}).collect();
let addresses = (0..100)
.map(|_| {
let peer_id = PeerId::random();
let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
address.with(multiaddr::Protocol::P2p(peer_id.into()))
})
.collect();
let dht_event = block_on(build_dht_event(
addresses,
remote_public.into(),
&remote_key_store,
));
let dht_event = block_on(build_dht_event(addresses, remote_public.into(), &remote_key_store));
let (_dht_event_tx, dht_event_rx) = channel(1);
@@ -506,16 +508,20 @@ fn limit_number_of_addresses_added_to_cache_per_authority() {
worker.handle_dht_value_found_event(vec![dht_event]).unwrap();
assert_eq!(
MAX_ADDRESSES_PER_AUTHORITY,
worker.addr_cache.get_addresses_by_authority_id(&remote_public.into()).unwrap().len(),
worker
.addr_cache
.get_addresses_by_authority_id(&remote_public.into())
.unwrap()
.len(),
);
}
#[test]
fn do_not_cache_addresses_without_peer_id() {
let remote_key_store = KeyStore::new();
let remote_public = block_on(remote_key_store
.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap();
let remote_public =
block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap();
let multiaddr_with_peer_id = {
let peer_id = PeerId::random();
@@ -524,21 +530,17 @@ fn do_not_cache_addresses_without_peer_id() {
address.with(multiaddr::Protocol::P2p(peer_id.into()))
};
let multiaddr_without_peer_id: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
let multiaddr_without_peer_id: Multiaddr =
"/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
let dht_event = block_on(build_dht_event(
vec![
multiaddr_with_peer_id.clone(),
multiaddr_without_peer_id,
],
vec![multiaddr_with_peer_id.clone(), multiaddr_without_peer_id],
remote_public.into(),
&remote_key_store,
));
let (_dht_event_tx, dht_event_rx) = channel(1);
let local_test_api = Arc::new(TestApi {
authorities: vec![remote_public.into()],
});
let local_test_api = Arc::new(TestApi { authorities: vec![remote_public.into()] });
let local_network: Arc<TestNetwork> = Arc::new(Default::default());
let local_key_store = KeyStore::new();
@@ -578,9 +580,7 @@ fn addresses_to_publish_adds_p2p() {
let (_to_worker, from_service) = mpsc::channel(0);
let worker = Worker::new(
from_service,
Arc::new(TestApi {
authorities: vec![],
}),
Arc::new(TestApi { authorities: vec![] }),
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(Arc::new(KeyStore::new())),
@@ -605,17 +605,16 @@ fn addresses_to_publish_respects_existing_p2p_protocol() {
let network: Arc<TestNetwork> = Arc::new(TestNetwork {
external_addresses: vec![
"/ip6/2001:db8::/tcp/30333/p2p/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSupNKC"
.parse().unwrap(),
.parse()
.unwrap(),
],
.. Default::default()
..Default::default()
});
let (_to_worker, from_service) = mpsc::channel(0);
let worker = Worker::new(
from_service,
Arc::new(TestApi {
authorities: vec![],
}),
Arc::new(TestApi { authorities: vec![] }),
network.clone(),
Box::pin(dht_event_rx),
Role::PublishAndDiscover(Arc::new(KeyStore::new())),
@@ -624,7 +623,8 @@ fn addresses_to_publish_respects_existing_p2p_protocol() {
);
assert_eq!(
network.external_addresses, worker.addresses_to_publish().collect::<Vec<_>>(),
network.external_addresses,
worker.addresses_to_publish().collect::<Vec<_>>(),
"Expected Multiaddr from `TestNetwork` to not be altered.",
);
}
@@ -635,21 +635,21 @@ fn lookup_throttling() {
let peer_id = PeerId::random();
let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
address.with(multiaddr::Protocol::P2p(
peer_id.into(),
))
address.with(multiaddr::Protocol::P2p(peer_id.into()))
};
let remote_key_store = KeyStore::new();
let remote_public_keys: Vec<AuthorityId> = (0..20).map(|_| {
block_on(remote_key_store
.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap().into()
}).collect();
let remote_hash_to_key = remote_public_keys.iter()
let remote_public_keys: Vec<AuthorityId> = (0..20)
.map(|_| {
block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
.unwrap()
.into()
})
.collect();
let remote_hash_to_key = remote_public_keys
.iter()
.map(|k| (hash_authority_id(k.as_ref()), k.clone()))
.collect::<HashMap<_, _>>();
let (mut dht_event_tx, dht_event_rx) = channel(1);
let (_to_worker, from_service) = mpsc::channel(0);
let mut network = TestNetwork::default();
@@ -668,56 +668,61 @@ fn lookup_throttling() {
let mut pool = LocalPool::new();
let metrics = worker.metrics.clone().unwrap();
let _ = pool.spawner().spawn_local_obj(async move {
// Refilling `pending_lookups` only happens every X minutes. Fast
// forward by calling `refill_pending_lookups_queue` directly.
worker.refill_pending_lookups_queue().await.unwrap();
worker.run().await
}.boxed_local().into());
pool.run_until(async {
// Assert worker to trigger MAX_IN_FLIGHT_LOOKUPS lookups.
for _ in 0..MAX_IN_FLIGHT_LOOKUPS {
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
let _ = pool.spawner().spawn_local_obj(
async move {
// Refilling `pending_lookups` only happens every X minutes. Fast
// forward by calling `refill_pending_lookups_queue` directly.
worker.refill_pending_lookups_queue().await.unwrap();
worker.run().await
}
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
.boxed_local()
.into(),
);
// Make first lookup succeed.
let remote_hash = network.get_value_call.lock().unwrap().pop().unwrap();
let remote_key: AuthorityId = remote_hash_to_key.get(&remote_hash).unwrap().clone();
let dht_event = {
let (key, value) = build_dht_event(
vec![remote_multiaddr.clone()],
remote_key,
&remote_key_store
).await;
sc_network::DhtEvent::ValueFound(vec![(key, value)])
};
dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
pool.run_until(
async {
// Assert worker to trigger MAX_IN_FLIGHT_LOOKUPS lookups.
for _ in 0..MAX_IN_FLIGHT_LOOKUPS {
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
}
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
// Assert worker to trigger another lookup.
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS - 1) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
// Make first lookup succeed.
let remote_hash = network.get_value_call.lock().unwrap().pop().unwrap();
let remote_key: AuthorityId = remote_hash_to_key.get(&remote_hash).unwrap().clone();
let dht_event = {
let (key, value) =
build_dht_event(vec![remote_multiaddr.clone()], remote_key, &remote_key_store)
.await;
sc_network::DhtEvent::ValueFound(vec![(key, value)])
};
dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
// Make second one fail.
let remote_hash = network.get_value_call.lock().unwrap().pop().unwrap();
let dht_event = sc_network::DhtEvent::ValueNotFound(remote_hash);
dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
// Assert worker to trigger another lookup.
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS - 1) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
// Assert worker to trigger another lookup.
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS - 2) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
}.boxed_local());
// Make second one fail.
let remote_hash = network.get_value_call.lock().unwrap().pop().unwrap();
let dht_event = sc_network::DhtEvent::ValueNotFound(remote_hash);
dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
// Assert worker to trigger another lookup.
assert!(matches!(receiver.next().await, Some(TestNetworkEvent::GetCalled(_))));
assert_eq!(
metrics.requests_pending.get(),
(remote_public_keys.len() - MAX_IN_FLIGHT_LOOKUPS - 2) as u64
);
assert_eq!(network.get_value_call.lock().unwrap().len(), MAX_IN_FLIGHT_LOOKUPS);
}
.boxed_local(),
);
}