Light client integration tests (#2638)

* add light nodes to test_sync && test_consensus

* forbid light2light connections

* add light nodes to test_connectivity

* fixed light2light connection penalty

* post-merge fixes

* remove best_queued from Client info

substrate/core/client/src/client.rs

@@ -168,10 +168,6 @@ pub trait BlockBody<Block: BlockT> {
 pub struct ClientInfo<Block: BlockT> {
 	/// Best block hash.
 	pub chain: ChainInfo<Block>,
-	/// Best block number in the queue.
-	pub best_queued_number: Option<<<Block as BlockT>::Header as HeaderT>::Number>,
-	/// Best queued block hash.
-	pub best_queued_hash: Option<Block::Hash>,
 }
 
 /// Block status.
@@ -1171,8 +1167,6 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
 		let info = self.backend.blockchain().info();
 		ClientInfo {
 			chain: info,
-			best_queued_hash: None,
-			best_queued_number: None,
 		}
 	}
 

substrate/core/network/src/protocol.rs

@@ -845,13 +845,22 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 				network_out.disconnect_peer(who);
 				return;
 			}
+
 			if self.config.roles.is_light() {
+				// we're not interested in light peers
+				if status.roles.is_light() {
+					debug!(target: "sync", "Peer {} is unable to serve light requests", who);
+					network_out.report_peer(who.clone(), i32::min_value());
+					network_out.disconnect_peer(who);
+					return;
+				}
+
+				// we don't interested in peers that are far behind us
 				let self_best_block = self
 					.context_data
 					.chain
 					.info()
-					.best_queued_number
-					.unwrap_or_else(|| Zero::zero());
+					.chain.best_number;
 				let blocks_difference = self_best_block
 					.checked_sub(&status.best_number)
 					.unwrap_or_else(Zero::zero)

substrate/core/network/src/sync.rs

@@ -191,8 +191,8 @@ impl<B: BlockT> ChainSync<B> {
 			_genesis_hash: info.chain.genesis_hash,
 			peers: HashMap::new(),
 			blocks: BlockCollection::new(),
-			best_queued_hash: info.best_queued_hash.unwrap_or(info.chain.best_hash),
-			best_queued_number: info.best_queued_number.unwrap_or(info.chain.best_number),
+			best_queued_hash: info.chain.best_hash,
+			best_queued_number: info.chain.best_number,
 			extra_requests: ExtraRequestsAggregator::new(),
 			role,
 			required_block_attributes,
@@ -852,8 +852,8 @@ impl<B: BlockT> ChainSync<B> {
 		self.best_importing_number = Zero::zero();
 		self.blocks.clear();
 		let info = protocol.client().info();
-		self.best_queued_hash = info.best_queued_hash.unwrap_or(info.chain.best_hash);
-		self.best_queued_number = info.best_queued_number.unwrap_or(info.chain.best_number);
+		self.best_queued_hash = info.chain.best_hash;
+		self.best_queued_number = info.chain.best_number;
 		debug!(target:"sync", "Restarted with {} ({})", self.best_queued_number, self.best_queued_hash);
 		let ids: Vec<PeerId> = self.peers.drain().map(|(id, _)| id).collect();
 		for id in ids {

substrate/core/service/test/src/lib.rs

@@ -46,23 +46,38 @@ struct TestNet<F: ServiceFactory> {
 	runtime: Runtime,
 	authority_nodes: Vec<(u32, Arc<F::FullService>, Multiaddr)>,
 	full_nodes: Vec<(u32, Arc<F::FullService>, Multiaddr)>,
-	_light_nodes: Vec<(u32, Arc<F::LightService>)>,
+	light_nodes: Vec<(u32, Arc<F::LightService>, Multiaddr)>,
 	chain_spec: FactoryChainSpec<F>,
 	base_port: u16,
 	nodes: usize,
 }
 
 impl<F: ServiceFactory> TestNet<F> {
-	pub fn run_until_all_full<P: Send + Sync + Fn(u32, &F::FullService) -> bool + 'static>(&mut self, predicate: P) {
+	pub fn run_until_all_full<FP, LP>(
+		&mut self,
+		full_predicate: FP,
+		light_predicate: LP,
+	)
+		where
+			FP: Send + Sync + Fn(u32, &F::FullService) -> bool + 'static,
+			LP: Send + Sync + Fn(u32, &F::LightService) -> bool + 'static,
+	{
 		let full_nodes = self.full_nodes.clone();
+		let light_nodes = self.light_nodes.clone();
 		let interval = Interval::new_interval(Duration::from_millis(100))
 			.map_err(|_| ())
 			.for_each(move |_| {
-				if full_nodes.iter().all(|&(ref id, ref service, _)| predicate(*id, service)) {
-					Err(())
-				} else {
-					Ok(())
+				let full_ready = full_nodes.iter().all(|&(ref id, ref service, _)| full_predicate(*id, service));
+				if !full_ready {
+					return Ok(());
 				}
+
+				let light_ready = light_nodes.iter().all(|&(ref id, ref service, _)| light_predicate(*id, service));
+				if !light_ready {
+					return Ok(());
+				}
+
+				Err(())
 			})
 			.timeout(MAX_WAIT_TIME);
 
@@ -152,7 +167,7 @@ impl<F: ServiceFactory> TestNet<F> {
 			runtime,
 			authority_nodes: Default::default(),
 			full_nodes: Default::default(),
-			_light_nodes: Default::default(),
+			light_nodes: Default::default(),
 			chain_spec: spec.clone(),
 			base_port,
 			nodes: 0,
@@ -186,28 +201,46 @@ impl<F: ServiceFactory> TestNet<F> {
 		}));
 		nodes += full as usize;
 
-		self._light_nodes.extend((nodes..nodes + light as usize).map(|index| (index as u32,
-			Arc::new(F::new_light(node_config::<F>(index as u32, &spec, Roles::LIGHT, None, base_port, &temp), executor.clone())
-					 .expect("Error creating test node service")))
-		));
+		self.light_nodes.extend((nodes..nodes + light as usize).map(|index| {
+			let node_config = node_config::<F>(index as u32, &spec, Roles::LIGHT, None, base_port, &temp);
+			let addr = node_config.network.listen_addresses.iter().next().unwrap().clone();
+			let service = Arc::new(F::new_light(node_config, executor.clone())
+				.expect("Error creating test node service"));
+			let addr = addr.with(multiaddr::Protocol::P2p(service.network().local_peer_id().into()));
+			(index as u32, service, addr)
+		}));
 		nodes += light as usize;
+
 		self.nodes = nodes;
 	}
 }
 
 pub fn connectivity<F: ServiceFactory>(spec: FactoryChainSpec<F>) {
-	const NUM_NODES: u32 = 10;
+	const NUM_FULL_NODES: u32 = 5;
+	const NUM_LIGHT_NODES: u32 = 5;
 	{
 		let temp = TempDir::new("substrate-connectivity-test").expect("Error creating test dir");
 		let runtime = {
-			let mut network = TestNet::<F>::new(&temp, spec.clone(), NUM_NODES, 0, vec![], 30400);
+			let mut network = TestNet::<F>::new(
+				&temp,
+				spec.clone(),
+				NUM_FULL_NODES,
+				NUM_LIGHT_NODES,
+				vec![],
+				30400,
+			);
 			info!("Checking star topology");
 			let first_address = network.full_nodes[0].2.clone();
 			for (_, service, _) in network.full_nodes.iter().skip(1) {
 				service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
 			}
-			network.run_until_all_full(|_index, service|
-				service.network().peers_debug_info().len() == NUM_NODES as usize - 1
+			for (_, service, _) in network.light_nodes.iter() {
+				service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
+			}
+			network.run_until_all_full(
+				|_index, service| service.network().peers_debug_info().len() == NUM_FULL_NODES as usize - 1
+					+ NUM_LIGHT_NODES as usize,
+				|_index, service| service.network().peers_debug_info().len() == NUM_FULL_NODES as usize,
 			);
 			network.runtime
 		};
@@ -219,16 +252,35 @@ pub fn connectivity<F: ServiceFactory>(spec: FactoryChainSpec<F>) {
 	{
 		let temp = TempDir::new("substrate-connectivity-test").expect("Error creating test dir");
 		{
-			let mut network = TestNet::<F>::new(&temp, spec, NUM_NODES, 0, vec![], 30400);
+			let mut network = TestNet::<F>::new(
+				&temp,
+				spec,
+				NUM_FULL_NODES,
+				NUM_LIGHT_NODES,
+				vec![],
+				30400,
+			);
 			info!("Checking linked topology");
 			let mut address = network.full_nodes[0].2.clone();
-			for (_, service, node_id) in network.full_nodes.iter().skip(1) {
-				service.network().add_reserved_peer(address.to_string()).expect("Error adding reserved peer");
-				address = node_id.clone();
+			let max_nodes = ::std::cmp::max(NUM_FULL_NODES, NUM_LIGHT_NODES);
+			for i in 0..max_nodes {
+				if i != 0 {
+					if let Some((_, service, node_id)) = network.full_nodes.get(i as usize) {
+						service.network().add_reserved_peer(address.to_string()).expect("Error adding reserved peer");
+						address = node_id.clone();
+					}
+				}
+
+				if let Some((_, service, node_id)) = network.light_nodes.get(i as usize) {
+					service.network().add_reserved_peer(address.to_string()).expect("Error adding reserved peer");
+					address = node_id.clone();
+				}
 			}
-			network.run_until_all_full(|_index, service| {
-				service.network().peers_debug_info().len() == NUM_NODES as usize - 1
-			});
+			network.run_until_all_full(
+				|_index, service| service.network().peers_debug_info().len() == NUM_FULL_NODES as usize - 1
+					+ NUM_LIGHT_NODES as usize,
+				|_index, service| service.network().peers_debug_info().len() == NUM_FULL_NODES as usize,
+			);
 		}
 		temp.close().expect("Error removing temp dir");
 	}
@@ -244,10 +296,18 @@ where
 	B: FnMut(&F::FullService) -> ImportBlock<F::Block>,
 	E: FnMut(&F::FullService) -> FactoryExtrinsic<F>,
 {
-	const NUM_NODES: u32 = 10;
+	const NUM_FULL_NODES: u32 = 10;
+	const NUM_LIGHT_NODES: u32 = 10;
 	const NUM_BLOCKS: u32 = 512;
 	let temp = TempDir::new("substrate-sync-test").expect("Error creating test dir");
-	let mut network = TestNet::<F>::new(&temp, spec.clone(), NUM_NODES, 0, vec![], 30500);
+	let mut network = TestNet::<F>::new(
+		&temp,
+		spec.clone(),
+		NUM_FULL_NODES,
+		NUM_LIGHT_NODES,
+		vec![],
+		30500,
+	);
 	info!("Checking block sync");
 	let first_address = {
 		let first_service = &network.full_nodes[0].1;
@@ -264,15 +324,22 @@ where
 	for (_, service, _) in network.full_nodes.iter().skip(1) {
 		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
 	}
-	network.run_until_all_full(|_index, service|
-		service.client().info().chain.best_number == NUM_BLOCKS.into()
+	for (_, service, _) in network.light_nodes.iter() {
+		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
+	}
+	network.run_until_all_full(
+		|_index, service|
+			service.client().info().chain.best_number == NUM_BLOCKS.into(),
+		|_index, service|
+			service.client().info().chain.best_number == NUM_BLOCKS.into(),
 	);
 	info!("Checking extrinsic propagation");
 	let first_service = network.full_nodes[0].1.clone();
 	let best_block = BlockId::number(first_service.client().info().chain.best_number);
 	first_service.transaction_pool().submit_one(&best_block, extrinsic_factory(&first_service)).unwrap();
-	network.run_until_all_full(|_index, service|
-		service.transaction_pool().ready().count() == 1
+	network.run_until_all_full(
+		|_index, service| service.transaction_pool().ready().count() == 1,
+		|_index, _service| true,
 	);
 }
 
@@ -280,27 +347,47 @@ pub fn consensus<F>(spec: FactoryChainSpec<F>, authorities: Vec<String>)
 	where
 		F: ServiceFactory,
 {
-	const NUM_NODES: u32 = 10;
+	const NUM_FULL_NODES: u32 = 10;
+	const NUM_LIGHT_NODES: u32 = 0;
 	const NUM_BLOCKS: u32 = 10; // 10 * 2 sec block production time = ~20 seconds
 	let temp = TempDir::new("substrate-conensus-test").expect("Error creating test dir");
-	let mut network = TestNet::<F>::new(&temp, spec.clone(), NUM_NODES / 2, 0, authorities, 30600);
+	let mut network = TestNet::<F>::new(
+		&temp,
+		spec.clone(),
+		NUM_FULL_NODES / 2,
+		NUM_LIGHT_NODES / 2,
+		authorities,
+		30600,
+	);
 	info!("Checking consensus");
 	let first_address = network.authority_nodes[0].2.clone();
 	for (_, service, _) in network.full_nodes.iter() {
 		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
 	}
+	for (_, service, _) in network.light_nodes.iter() {
+		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
+	}
 	for (_, service, _) in network.authority_nodes.iter().skip(1) {
 		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
 	}
-	network.run_until_all_full(|_index, service| {
-		service.client().info().chain.finalized_number >= (NUM_BLOCKS / 2).into()
-	});
+	network.run_until_all_full(
+		|_index, service|
+			service.client().info().chain.finalized_number >= (NUM_BLOCKS / 2).into(),
+		|_index, service|
+			service.client().info().chain.best_number >= (NUM_BLOCKS / 2).into(),
+	);
 	info!("Adding more peers");
-	network.insert_nodes(&temp, NUM_NODES / 2, 0, vec![]);
+	network.insert_nodes(&temp, NUM_FULL_NODES / 2, NUM_LIGHT_NODES / 2, vec![]);
 	for (_, service, _) in network.full_nodes.iter() {
 		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
 	}
-	network.run_until_all_full(|_index, service|
-		service.client().info().chain.finalized_number >= NUM_BLOCKS.into()
+	for (_, service, _) in network.light_nodes.iter() {
+		service.network().add_reserved_peer(first_address.to_string()).expect("Error adding reserved peer");
+	}
+	network.run_until_all_full(
+		|_index, service|
+			service.client().info().chain.finalized_number >= NUM_BLOCKS.into(),
+		|_index, service|
+			service.client().info().chain.best_number >= NUM_BLOCKS.into(),
 	);
 }