client/beefy: request justifs from peers further in consensus (#13343)

For on-demand justifications, peer selection is based on witnessed
gossip votes. This commit changes the condition for selecting a peer
to request justification for `block` from
"last voted on >= `block`" to "peer last voted on strict > `block`".

When allowing `>=` we see nodes continuously spamming unsuccessful
on-demand requests to nodes which are still voting on a block without
having a justification available.

One way to fix the spam would be to add some rate-limiting or backoff
period when requesting justifications.

The other solution (present in this commit) is to simply request
justifications from peers that are voting on future blocks so we know
they're _guaranteed_ to have the wanted mandatory justification
available to send back.

Signed-off-by: acatangiu <adrian@parity.io>

substrate/client/beefy/src/communication/peers.rs

@@ -57,11 +57,11 @@ impl<B: Block> KnownPeers<B> {
 		self.live.remove(peer);
 	}
 
-	/// Return _filtered and cloned_ list of peers that have voted on `block` or higher.
-	pub fn at_least_at_block(&self, block: NumberFor<B>) -> VecDeque<PeerId> {
+	/// Return _filtered and cloned_ list of peers that have voted on higher than `block`.
+	pub fn further_than(&self, block: NumberFor<B>) -> VecDeque<PeerId> {
 		self.live
 			.iter()
-			.filter_map(|(k, v)| (v.last_voted_on >= block).then_some(k))
+			.filter_map(|(k, v)| (v.last_voted_on > block).then_some(k))
 			.cloned()
 			.collect()
 	}
@@ -92,22 +92,22 @@ mod tests {
 		assert!(peers.contains(&bob));
 		assert!(peers.contains(&charlie));
 
-		// Get peers at block >= 5
-		let at_5 = peers.at_least_at_block(5);
+		// Get peers at block > 4
+		let further_than_4 = peers.further_than(4);
 		// Should be Bob and Charlie
-		assert_eq!(at_5.len(), 2);
-		assert!(at_5.contains(&bob));
-		assert!(at_5.contains(&charlie));
+		assert_eq!(further_than_4.len(), 2);
+		assert!(further_than_4.contains(&bob));
+		assert!(further_than_4.contains(&charlie));
 
 		// 'Tracked' Alice seen voting for 10.
 		peers.note_vote_for(alice, 10);
 
-		// Get peers at block >= 9
-		let at_9 = peers.at_least_at_block(9);
+		// Get peers at block > 9
+		let further_than_9 = peers.further_than(9);
 		// Should be Charlie and Alice
-		assert_eq!(at_9.len(), 2);
-		assert!(at_9.contains(&charlie));
-		assert!(at_9.contains(&alice));
+		assert_eq!(further_than_9.len(), 2);
+		assert!(further_than_9.contains(&charlie));
+		assert!(further_than_9.contains(&alice));
 
 		// Remove Alice
 		peers.remove(&alice);
@@ -115,9 +115,9 @@ mod tests {
 		assert!(!peers.contains(&alice));
 
 		// Get peers at block >= 9
-		let at_9 = peers.at_least_at_block(9);
+		let further_than_9 = peers.further_than(9);
 		// Now should be just Charlie
-		assert_eq!(at_9.len(), 1);
-		assert!(at_9.contains(&charlie));
+		assert_eq!(further_than_9.len(), 1);
+		assert!(further_than_9.contains(&charlie));
 	}
 }

substrate/client/beefy/src/communication/request_response/outgoing_requests_engine.rs

@@ -80,7 +80,7 @@ impl<B: Block> OnDemandJustificationsEngine<B> {
 
 	fn reset_peers_cache_for_block(&mut self, block: NumberFor<B>) {
 		// TODO (issue #12296): replace peer selection with generic one that involves all protocols.
-		self.peers_cache = self.live_peers.lock().at_least_at_block(block);
+		self.peers_cache = self.live_peers.lock().further_than(block);
 	}
 
 	fn try_next_peer(&mut self) -> Option<PeerId> {
@@ -199,7 +199,6 @@ impl<B: Block> OnDemandJustificationsEngine<B> {
 		let (peer, req_info, resp) = match &mut self.state {
 			State::Idle => {
 				futures::pending!();
-				// Doesn't happen as 'futures::pending!()' is an 'await' barrier that never passes.
 				return None
 			},
 			State::AwaitingResponse(peer, req_info, receiver) => {

substrate/client/beefy/src/tests.rs

@@ -893,7 +893,7 @@ async fn on_demand_beefy_justification_sync() {
 		[BeefyKeyring::Alice, BeefyKeyring::Bob, BeefyKeyring::Charlie, BeefyKeyring::Dave];
 	let validator_set = ValidatorSet::new(make_beefy_ids(&all_peers), 0).unwrap();
 	let session_len = 5;
-	let min_block_delta = 5;
+	let min_block_delta = 4;
 
 	let mut net = BeefyTestNet::new(4);
 
@@ -912,7 +912,7 @@ async fn on_demand_beefy_justification_sync() {
 	let dave_index = 3;
 
 	// push 30 blocks
-	let hashes = net.generate_blocks_and_sync(30, session_len, &validator_set, false).await;
+	let hashes = net.generate_blocks_and_sync(35, session_len, &validator_set, false).await;
 
 	let fast_peers = fast_peers.into_iter().enumerate();
 	let net = Arc::new(Mutex::new(net));
@@ -921,7 +921,7 @@ async fn on_demand_beefy_justification_sync() {
 	finalize_block_and_wait_for_beefy(
 		&net,
 		fast_peers.clone(),
-		&[hashes[1], hashes[6], hashes[10], hashes[17], hashes[24]],
+		&[hashes[1], hashes[6], hashes[10], hashes[17], hashes[23]],
 		&[1, 5, 10, 15, 20],
 	)
 	.await;
@@ -941,12 +941,12 @@ async fn on_demand_beefy_justification_sync() {
 	// freshly spun up Dave now needs to listen for gossip to figure out the state of their peers.
 
 	// Have the other peers do some gossip so Dave finds out about their progress.
-	finalize_block_and_wait_for_beefy(&net, fast_peers, &[hashes[25]], &[25]).await;
+	finalize_block_and_wait_for_beefy(&net, fast_peers, &[hashes[25], hashes[29]], &[25, 29]).await;
 
 	// Now verify Dave successfully finalized #1 (through on-demand justification request).
 	wait_for_best_beefy_blocks(dave_best_blocks, &net, &[1]).await;
 
-	// Give Dave all tasks some cpu cycles to burn through their events queues,
+	// Give all tasks some cpu cycles to burn through their events queues,
 	run_for(Duration::from_millis(100), &net).await;
 	// then verify Dave catches up through on-demand justification requests.
 	finalize_block_and_wait_for_beefy(
@@ -959,7 +959,7 @@ async fn on_demand_beefy_justification_sync() {
 
 	let all_peers = all_peers.into_iter().enumerate();
 	// Now that Dave has caught up, sanity check voting works for all of them.
-	finalize_block_and_wait_for_beefy(&net, all_peers, &[hashes[30]], &[30]).await;
+	finalize_block_and_wait_for_beefy(&net, all_peers, &[hashes[30], hashes[34]], &[30]).await;
 }
 
 #[tokio::test]

substrate/client/beefy/src/worker.rs

@@ -555,7 +555,6 @@ where
 		let block_number = vote.commitment.block_number;
 		match rounds.add_vote(vote) {
 			VoteImportResult::RoundConcluded(signed_commitment) => {
-				self.gossip_validator.conclude_round(block_number);
 				metric_set!(self, beefy_round_concluded, block_number);
 
 				let finality_proof = VersionedFinalityProof::V1(signed_commitment);
@@ -602,6 +601,7 @@ where
 
 		// Finalize inner round and update voting_oracle state.
 		self.persisted_state.voting_oracle.finalize(block_num)?;
+		self.gossip_validator.conclude_round(block_num);
 
 		if block_num > self.best_beefy_block() {
 			// Set new best BEEFY block number.