client/beefy: request justifs from peers further in consensus

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,32 +92,32 @@ 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);
 		assert_eq!(peers.live.len(), 2);
 		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));
 	}
 }

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) => {

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]

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.