sync: process empty response for justification requests (#4957)

* sync: process empty response for justification request

* sync: add test for justification request empty response

* network: remove deprecated comment

substrate/client/network/src/protocol.rs

@@ -864,8 +864,6 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			blocks_range
 		);
 
-		// TODO [andre]: move this logic to the import queue so that
-		// justifications are imported asynchronously (#1482)
 		if request.fields == message::BlockAttributes::JUSTIFICATION {
 			match self.sync.on_block_justification(peer, response) {
 				Ok(sync::OnBlockJustification::Nothing) => CustomMessageOutcome::None,

substrate/client/network/src/protocol/sync.rs

@@ -813,7 +813,7 @@ impl<B: BlockT> ChainSync<B> {
 			peer.state = PeerSyncState::Available;
 
 			// We only request one justification at a time
-			if let Some(block) = response.blocks.into_iter().next() {
+			let justification = if let Some(block) = response.blocks.into_iter().next() {
 				if hash != block.hash {
 					info!(
 						target: "sync",
@@ -821,13 +821,22 @@ impl<B: BlockT> ChainSync<B> {
 					);
 					return Err(BadPeer(who, rep::BAD_JUSTIFICATION));
 				}
-				if let Some((peer, hash, number, j)) = self.extra_justifications.on_response(who, block.justification) {
-					return Ok(OnBlockJustification::Import { peer, hash, number, justification: j })
-				}
+
+				block.justification
 			} else {
-				// we might have asked the peer for a justification on a block that we thought it had
-				// (regardless of whether it had a justification for it or not).
-				trace!(target: "sync", "Peer {:?} provided empty response for justification request {:?}", who, hash)
+				// we might have asked the peer for a justification on a block that we assumed it
+				// had but didn't (regardless of whether it had a justification for it or not).
+				trace!(target: "sync",
+					"Peer {:?} provided empty response for justification request {:?}",
+					who,
+					hash,
+				);
+
+				None
+			};
+
+			if let Some((peer, hash, number, j)) = self.extra_justifications.on_response(who, justification) {
+				return Ok(OnBlockJustification::Import { peer, hash, number, justification: j })
 			}
 		}
 
@@ -1353,3 +1362,93 @@ fn fork_sync_request<B: BlockT>(
 	}
 	None
 }
+
+#[cfg(test)]
+mod test {
+	use super::*;
+	use super::message::FromBlock;
+	use substrate_test_runtime_client::{
+		runtime::Block,
+		DefaultTestClientBuilderExt, TestClientBuilder, TestClientBuilderExt,
+	};
+	use sp_blockchain::HeaderBackend;
+	use sp_consensus::block_validation::DefaultBlockAnnounceValidator;
+
+	#[test]
+	fn processes_empty_response_on_justification_request_for_unknown_block() {
+		// if we ask for a justification for a given block to a peer that doesn't know that block
+		// (different from not having a justification), the peer will reply with an empty response.
+		// internally we should process the response as the justification not being available.
+
+		let client = Arc::new(TestClientBuilder::new().build());
+		let info = client.info();
+		let block_announce_validator = Box::new(DefaultBlockAnnounceValidator::new(client.clone()));
+		let peer_id = PeerId::random();
+
+		let mut sync = ChainSync::new(
+			Roles::AUTHORITY,
+			client.clone(),
+			&info,
+			None,
+			block_announce_validator,
+			1,
+		);
+
+		let (a1_hash, a1_number) = {
+			let a1 = client.new_block(Default::default()).unwrap().build().unwrap().block;
+			(a1.hash(), *a1.header.number())
+		};
+
+		// add a new peer with the same best block
+		sync.new_peer(peer_id.clone(), a1_hash, a1_number).unwrap();
+
+		// and request a justification for the block
+		sync.request_justification(&a1_hash, a1_number);
+
+		// the justification request should be scheduled to that peer
+		assert!(
+			sync.justification_requests().any(|(who, request)| {
+				who == peer_id && request.from == FromBlock::Hash(a1_hash)
+			})
+		);
+
+		// there are no extra pending requests
+		assert_eq!(
+			sync.extra_justifications.pending_requests().count(),
+			0,
+		);
+
+		// there's one in-flight extra request to the expected peer
+		assert!(
+			sync.extra_justifications.active_requests().any(|(who, (hash, number))| {
+				*who == peer_id && *hash == a1_hash && *number == a1_number
+			})
+		);
+
+		// if the peer replies with an empty response (i.e. it doesn't know the block),
+		// the active request should be cleared.
+		assert_eq!(
+			sync.on_block_justification(
+				peer_id.clone(),
+				BlockResponse::<Block> {
+					id: 0,
+					blocks: vec![],
+				}
+			),
+			Ok(OnBlockJustification::Nothing),
+		);
+
+		// there should be no in-flight requests
+		assert_eq!(
+			sync.extra_justifications.active_requests().count(),
+			0,
+		);
+
+		// and the request should now be pending again, waiting for reschedule
+		assert!(
+			sync.extra_justifications.pending_requests().any(|(hash, number)| {
+				*hash == a1_hash && *number == a1_number
+			})
+		);
+	}
+}

substrate/client/network/src/protocol/sync/extra_requests.rs

@@ -228,6 +228,18 @@ impl<B: BlockT> ExtraRequests<B> {
 
 		true
 	}
+
+	/// Returns an iterator over all active (in-flight) requests and associated peer id.
+	#[cfg(test)]
+	pub(crate) fn active_requests(&self) -> impl Iterator<Item = (&PeerId, &ExtraRequest<B>)> {
+		self.active_requests.iter()
+	}
+
+	/// Returns an iterator over all scheduled pending requests.
+	#[cfg(test)]
+	pub(crate) fn pending_requests(&self) -> impl Iterator<Item = &ExtraRequest<B>> {
+		self.pending_requests.iter()
+	}
 }
 
 /// Matches peers with pending extra requests.