client/authority-discovery: Limit number of connections to authorities (#4487)

* client/authority-discovery: Limit number of connections to authorities

Instead of connecting to all sentry nodes of all authorities, with this
patch the authority discovery module does the following:

- Choose one sentry node per authority at random.

- Choose MAX_NUM_AUTHORITY_CONN out of the above at random.

The module uses randomness to prevent hot spots, e.g. all nodes trying
to connect to a single node. If the authority discovery module would
choose the nodes to connect to at random on each new address that it
learns of, the node would go through a lot of connection churn.  Instead
it creates a random seed at start up and uses this seed for its RNG on
each update cycle.

* client/authority-discovery: Extract address cache into own module

* client/authority-discovery/src/addr_cache: Add basic unit tests

* client/authority-discovery: Replace unwrap with expect on [u8] cmp

* .maintain/sentry-node/docker-compose.yml: Prefix endpoint flags

* client/authority-discovery/src/addr_cache: Use sort_unstable and cmp

* client/authority-discovery: Use BTreeMap in addr_cache for sorted iter

To reduce connection churn it is preferrable to have `get_subset` of the
`addr_cache` to return the same result on repeated calls. `get_subset`
iterates a map. To make the process of iteration deterministic, use a
`BTreeMap` instead of a `HashMap`.

substrate/client/authority-discovery/src/lib.rs

@@ -45,7 +45,6 @@
 //!    4. Adds the retrieved external addresses as priority nodes to the peerset.
 use std::collections::{HashMap, HashSet};
 use std::convert::TryInto;
-use std::iter::FromIterator;
 use std::marker::PhantomData;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -55,26 +54,26 @@ use futures::task::{Context, Poll};
 use futures::{Future, FutureExt, Stream, StreamExt};
 use futures_timer::Delay;
 
-use sp_authority_discovery::{
-	AuthorityDiscoveryApi, AuthorityId, AuthoritySignature, AuthorityPair
-};
-use sc_client_api::blockchain::HeaderBackend;
 use codec::{Decode, Encode};
 use error::{Error, Result};
-use log::{debug, error, log_enabled, warn};
 use libp2p::Multiaddr;
+use log::{debug, error, log_enabled, warn};
+use prost::Message;
+use sc_client_api::blockchain::HeaderBackend;
 use sc_network::specialization::NetworkSpecialization;
 use sc_network::{DhtEvent, ExHashT, NetworkStateInfo};
+use sp_authority_discovery::{AuthorityDiscoveryApi, AuthorityId, AuthoritySignature, AuthorityPair};
 use sp_core::crypto::{key_types, Pair};
 use sp_core::traits::BareCryptoStorePtr;
-use prost::Message;
 use sp_runtime::generic::BlockId;
 use sp_runtime::traits::{Block as BlockT, ProvideRuntimeApi};
+use addr_cache::AddrCache;
 
 #[cfg(test)]
 mod tests;
 
 mod error;
+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"));
@@ -89,12 +88,6 @@ const LIBP2P_KADEMLIA_BOOTSTRAP_TIME: Duration = Duration::from_secs(30);
 /// discovery module.
 const AUTHORITIES_PRIORITY_GROUP_NAME: &'static str = "authorities";
 
-/// The maximum number of sentry node public addresses that we accept per authority.
-///
-/// Everything above this threshold should be dropped to prevent a single authority from filling up
-/// our peer set priority group.
-const MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY: usize = 5;
-
 /// An `AuthorityDiscovery` makes a given authority discoverable and discovers other authorities.
 pub struct AuthorityDiscovery<Client, Network, Block>
 where
@@ -124,12 +117,7 @@ where
 	/// Interval on which to query for addresses of other authorities.
 	query_interval: Interval,
 
-	/// The network peerset interface for priority groups lets us only set an entire group, but we
-	/// retrieve the addresses of other authorities one by one from the network. To use the peerset
-	/// interface we need to cache the addresses and always overwrite the entire peerset priority
-	/// group. To ensure this map doesn't grow indefinitely `purge_old_authorities_from_cache`
-	/// function is called each time we add a new entry.
-	address_cache: HashMap<AuthorityId, Vec<Multiaddr>>,
+	addr_cache: addr_cache::AddrCache<AuthorityId, Multiaddr>,
 
 	phantom: PhantomData<Block>,
 }
@@ -182,22 +170,12 @@ where
 				}
 			}).collect::<Vec<Multiaddr>>();
 
-			if addrs.len() > MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY {
-				warn!(
-					target: "sub-authority-discovery",
-					"More than MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY ({:?}) were specified. Other \
-					nodes will likely ignore the remainder.",
-					MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY,
-				);
-			}
-
 			Some(addrs)
 		} else {
 			None
 		};
 
-
-		let address_cache = HashMap::new();
+		let addr_cache = AddrCache::new();
 
 		AuthorityDiscovery {
 			client,
@@ -207,7 +185,7 @@ where
 			key_store,
 			publish_interval,
 			query_interval,
-			address_cache,
+			addr_cache,
 			phantom: PhantomData,
 		}
 	}
@@ -304,86 +282,70 @@ where
 		&mut self,
 		values: Vec<(libp2p::kad::record::Key, Vec<u8>)>,
 	) -> Result<()> {
-		debug!(target: "sub-authority-discovery", "Got Dht value from network.");
-
-		let block_id = BlockId::hash(self.client.info().best_hash);
-
-		// From the Dht we only get the hashed authority id. In order to retrieve the actual
-		// authority id and to ensure it is actually an authority, we match the hash against the
-		// hash of the authority id of all other authorities.
-		let authorities = self.client.runtime_api().authorities(&block_id)?;
-		self.purge_old_authorities_from_cache(&authorities);
-
-		let authorities = authorities
-			.into_iter()
-			.map(|id| hash_authority_id(id.as_ref()).map(|h| (h, id)))
-			.collect::<Result<HashMap<_, _>>>()?;
-
-		for (key, value) in values.iter() {
-			// Check if the event origins from an authority in the current authority set.
-			let authority_id: &AuthorityId = authorities
-				.get(key)
-				.ok_or(Error::MatchingHashedAuthorityIdWithAuthorityId)?;
-
-			let schema::SignedAuthorityAddresses {
-				signature,
-				addresses,
-			} = schema::SignedAuthorityAddresses::decode(value).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);
+		// Ensure `values` is not empty and all its keys equal.
+		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
+				),
+				x @ Ok(_) => x,
+				Err(e) => Err(e),
 			}
+		})?.ok_or(Error::ReceivingDhtValueFoundEventWithNoRecords)?;
 
-			let mut addresses: Vec<libp2p::Multiaddr> = schema::AuthorityAddresses::decode(addresses)
-				.map(|a| a.addresses)
-				.map_err(Error::DecodingProto)?
+		let authorities = {
+			let block_id = BlockId::hash(self.client.info().best_hash);
+			// From the Dht we only get the hashed authority id. In order to retrieve the actual
+			// authority id and to ensure it is actually an authority, we match the hash against the
+			// hash of the authority id of all other authorities.
+			let authorities = self.client.runtime_api().authorities(&block_id)?;
+			self.addr_cache.retain_ids(&authorities);
+			authorities
 				.into_iter()
-				.map(|a| a.try_into())
-				.collect::<std::result::Result<_, _>>()
-				.map_err(Error::ParsingMultiaddress)?;
+				.map(|id| hash_authority_id(id.as_ref()).map(|h| (h, id)))
+				.collect::<Result<HashMap<_, _>>>()?
+		};
 
-			if addresses.len() > MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY {
-				warn!(
-					target: "sub-authority-discovery",
-					"Got more than MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY ({:?}) for Authority \
-					'{:?}' from DHT, dropping the remainder.",
-					MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY,	authority_id,
-				);
-				addresses = addresses.into_iter()
-					.take(MAX_NUM_SENTRY_ADDRESSES_PER_AUTHORITY)
-					.collect();
-			}
+		// Check if the event origins from an authority in the current authority set.
+		let authority_id: &AuthorityId = authorities
+			.get(&remote_key)
+			.ok_or(Error::MatchingHashedAuthorityIdWithAuthorityId)?;
 
-			self.address_cache.insert(authority_id.clone(), addresses);
+		let remote_addresses: Vec<Multiaddr> = values.into_iter()
+			.map(|(_k, v)| {
+				let schema::SignedAuthorityAddresses {
+					signature,
+					addresses,
+				} = schema::SignedAuthorityAddresses::decode(v).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);
+				}
+
+				let addresses: Vec<libp2p::Multiaddr> = schema::AuthorityAddresses::decode(addresses)
+					.map(|a| a.addresses)
+					.map_err(Error::DecodingProto)?
+					.into_iter()
+					.map(|a| a.try_into())
+					.collect::<std::result::Result<_, _>>()
+					.map_err(Error::ParsingMultiaddress)?;
+
+				Ok(addresses)
+			})
+			.collect::<Result<Vec<Vec<Multiaddr>>>>()?
+			.into_iter().flatten().collect();
+
+		if !remote_addresses.is_empty() {
+			self.addr_cache.insert(authority_id.clone(), remote_addresses);
+			self.update_peer_set_priority_group()?;
 		}
 
-		// Let's update the peerset priority group with all the addresses we have in our cache.
-
-		let addresses = HashSet::from_iter(
-			self.address_cache
-				.iter()
-				.map(|(_peer_id, addresses)| addresses.clone())
-				.flatten(),
-		);
-
-		debug!(
-			target: "sub-authority-discovery",
-			"Applying priority group {:#?} to peerset.", addresses,
-		);
-		self.network
-			.set_priority_group(AUTHORITIES_PRIORITY_GROUP_NAME.to_string(), addresses)
-			.map_err(Error::SettingPeersetPriorityGroup)?;
-
 		Ok(())
 	}
 
-	fn purge_old_authorities_from_cache(&mut self, current_authorities: &Vec<AuthorityId>) {
-		self.address_cache
-			.retain(|peer_id, _addresses| current_authorities.contains(peer_id))
-	}
-
 	/// Retrieve all local authority discovery private keys that are within the current authority
 	/// set.
 	fn get_priv_keys_within_authority_set(&mut self) -> Result<Vec<AuthorityPair>> {
@@ -429,6 +391,22 @@ where
 
 		Ok(intersection)
 	}
+
+	/// Update the peer set 'authority' priority group.
+	//
+	fn update_peer_set_priority_group(&self) -> Result<()>{
+		let addresses = self.addr_cache.get_subset();
+
+		debug!(
+			target: "sub-authority-discovery",
+			"Applying priority group {:?} to peerset.", addresses,
+		);
+		self.network
+			.set_priority_group(AUTHORITIES_PRIORITY_GROUP_NAME.to_string(), addresses.into_iter().collect())
+			.map_err(Error::SettingPeersetPriorityGroup)?;
+
+		Ok(())
+	}
 }
 
 impl<Client, Network, Block> Future for AuthorityDiscovery<Client, Network, Block>