Check for block parent before verification. (#1714)

* Treat verification errors more seriously

* Track obsolete requests

* Check block parent before verification

* Style

substrate/core/client/src/client.rs

@@ -1302,6 +1302,27 @@ impl<B, E, Block, RA> consensus::BlockImport<Block> for Client<B, E, Block, RA>
 			self.apply_block(operation, import_block, new_authorities)
 		}).map_err(|e| ConsensusErrorKind::ClientImport(e.to_string()).into())
 	}
+
+	/// Check block preconditions.
+	fn check_block(
+		&self,
+		hash: Block::Hash,
+		parent_hash: Block::Hash,
+	) -> Result<ImportResult, Self::Error> {
+		match self.backend.blockchain().status(BlockId::Hash(parent_hash))
+			.map_err(|e| ConsensusError::from(ConsensusErrorKind::ClientImport(e.to_string())))?
+		{
+			blockchain::BlockStatus::InChain => {},
+			blockchain::BlockStatus::Unknown => return Ok(ImportResult::UnknownParent),
+		}
+		match self.backend.blockchain().status(BlockId::Hash(hash))
+			.map_err(|e| ConsensusError::from(ConsensusErrorKind::ClientImport(e.to_string())))?
+		{
+			blockchain::BlockStatus::InChain => return Ok(ImportResult::AlreadyInChain),
+			blockchain::BlockStatus::Unknown => {},
+		}
+		Ok(ImportResult::Queued)
+	}
 }
 
 impl<B, E, Block, RA> consensus::Authorities<Block> for Client<B, E, Block, RA> where

substrate/core/consensus/common/src/block_import.rs

@@ -144,6 +144,13 @@ impl<Block: BlockT> ImportBlock<Block> {
 pub trait BlockImport<B: BlockT> {
 	type Error: ::std::error::Error + Send + 'static;
 
+	/// Check block preconditions.
+	fn check_block(
+		&self,
+		hash: B::Hash,
+		parent_hash: B::Hash,
+	) -> Result<ImportResult, Self::Error>;
+
 	/// Import a Block alongside the new authorities valid from this block forward
 	fn import_block(
 		&self,

substrate/core/consensus/common/src/import_queue.rs

@@ -408,6 +408,44 @@ pub fn import_single_block<B: BlockT, V: Verifier<B>>(
 	let number = header.number().clone();
 	let hash = header.hash();
 	let parent = header.parent_hash().clone();
+
+	let import_error = |e| {
+		match e {
+			Ok(ImportResult::AlreadyInChain) => {
+				trace!(target: "sync", "Block already in chain {}: {:?}", number, hash);
+				Ok(BlockImportResult::ImportedKnown(hash, number))
+			},
+			Ok(ImportResult::AlreadyQueued) => {
+				trace!(target: "sync", "Block already queued {}: {:?}", number, hash);
+				Ok(BlockImportResult::ImportedKnown(hash, number))
+			},
+			Ok(ImportResult::Queued) => {
+				Ok(BlockImportResult::ImportedUnknown(hash, number))
+			},
+			Ok(ImportResult::NeedsJustification) => {
+				trace!(target: "sync", "Block queued but requires justification {}: {:?}", number, hash);
+				Ok(BlockImportResult::ImportedUnjustified(hash, number))
+			},
+			Ok(ImportResult::UnknownParent) => {
+				debug!(target: "sync", "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent);
+				Err(BlockImportError::UnknownParent)
+			},
+			Ok(ImportResult::KnownBad) => {
+				debug!(target: "sync", "Peer gave us a bad block {}: {:?}", number, hash);
+				Err(BlockImportError::BadBlock(peer))
+			},
+			Err(e) => {
+				debug!(target: "sync", "Error importing block {}: {:?}: {:?}", number, hash, e);
+				Err(BlockImportError::Error)
+			}
+		}
+	};
+
+	match import_error(import_handle.check_block(hash, parent))? {
+		BlockImportResult::ImportedUnknown(_, _) => (),
+		r @ _ => return Ok(r), // Any other successfull result means that the block is already imported.
+	}
+
 	let (import_block, new_authorities) = verifier.verify(block_origin, header, justification, block.body)
 		.map_err(|msg| {
 			if let Some(peer) = peer {
@@ -418,36 +456,7 @@ pub fn import_single_block<B: BlockT, V: Verifier<B>>(
 			BlockImportError::VerificationFailed(peer, msg)
 		})?;
 
-	match import_handle.import_block(import_block, new_authorities) {
-		Ok(ImportResult::AlreadyInChain) => {
-			trace!(target: "sync", "Block already in chain {}: {:?}", number, hash);
-			Ok(BlockImportResult::ImportedKnown(hash, number))
-		},
-		Ok(ImportResult::AlreadyQueued) => {
-			trace!(target: "sync", "Block already queued {}: {:?}", number, hash);
-			Ok(BlockImportResult::ImportedKnown(hash, number))
-		},
-		Ok(ImportResult::Queued) => {
-			trace!(target: "sync", "Block queued {}: {:?}", number, hash);
-			Ok(BlockImportResult::ImportedUnknown(hash, number))
-		},
-		Ok(ImportResult::NeedsJustification) => {
-			trace!(target: "sync", "Block queued but requires justification {}: {:?}", number, hash);
-			Ok(BlockImportResult::ImportedUnjustified(hash, number))
-		},
-		Ok(ImportResult::UnknownParent) => {
-			debug!(target: "sync", "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent);
-			Err(BlockImportError::UnknownParent)
-		},
-		Ok(ImportResult::KnownBad) => {
-			debug!(target: "sync", "Peer gave us a bad block {}: {:?}", number, hash);
-			Err(BlockImportError::BadBlock(peer))
-		},
-		Err(e) => {
-			debug!(target: "sync", "Error importing block {}: {:?}: {:?}", number, hash, e);
-			Err(BlockImportError::Error)
-		}
-	}
+	import_error(import_handle.import_block(import_block, new_authorities))
 }
 
 /// Process single block import result.
@@ -472,13 +481,13 @@ pub fn process_import_result<B: BlockT>(
 		},
 		Err(BlockImportError::IncompleteHeader(who)) => {
 			if let Some(peer) = who {
-				link.useless_peer(peer, "Sent block with incomplete header to import");
+				link.note_useless_and_restart_sync(peer, "Sent block with incomplete header to import");
 			}
 			0
 		},
 		Err(BlockImportError::VerificationFailed(who, e)) => {
 			if let Some(peer) = who {
-				link.useless_peer(peer, &format!("Verification failed: {}", e));
+				link.note_useless_and_restart_sync(peer, &format!("Verification failed: {}", e));
 			}
 			0
 		},

substrate/core/finality-grandpa/src/import.rs

@@ -249,6 +249,14 @@ impl<B, E, Block: BlockT<Hash=H256>, RA, PRA> BlockImport<Block>
 
 		Ok(import_result)
 	}
+
+	fn check_block(
+		&self,
+		hash: Block::Hash,
+		parent_hash: Block::Hash,
+	) -> Result<ImportResult, Self::Error> {
+		self.inner.check_block(hash, parent_hash)
+	}
 }
 
 impl<B, E, Block: BlockT<Hash=H256>, RA, PRA> GrandpaBlockImport<B, E, Block, RA, PRA> {

substrate/core/network/src/protocol.rs

@@ -91,14 +91,10 @@ struct Peer<B: BlockT, H: ExHashT> {
 	best_hash: B::Hash,
 	/// Peer best block number
 	best_number: <B::Header as HeaderT>::Number,
-	/// Pending block request if any
-	block_request: Option<message::BlockRequest<B>>,
-	/// Pending block request timestamp
-	block_request_timestamp: Option<time::Instant>,
-	/// Pending block justification request if any
-	justification_request: Option<message::BlockRequest<B>>,
-	/// Pending block justification request timestamp
-	justification_request_timestamp: Option<time::Instant>,
+	/// Current block request, if any.
+	block_request: Option<(time::Instant, message::BlockRequest<B>)>,
+	/// Requests we are no longer insterested in.
+	obsolete_requests: HashMap<message::RequestId, time::Instant>,
 	/// Holds a set of transactions known to this peer.
 	known_extrinsics: HashSet<H>,
 	/// Holds a set of blocks known to this peer.
@@ -107,18 +103,6 @@ struct Peer<B: BlockT, H: ExHashT> {
 	next_request_id: message::RequestId,
 }
 
-impl<B: BlockT, H: ExHashT> Peer<B, H> {
-	fn min_request_timestamp(&self) -> Option<&time::Instant> {
-		match (self.block_request_timestamp, self.justification_request_timestamp) {
-			(Some(t1), Some(t2)) if t1 < t2 => self.block_request_timestamp.as_ref(),
-			(Some(_), Some(_)) => self.justification_request_timestamp.as_ref(),
-			(Some(_), None) => self.block_request_timestamp.as_ref(),
-			(None, Some(_)) => self.justification_request_timestamp.as_ref(),
-			_ => None,
-		}
-	}
-}
-
 /// Info about a peer's known state.
 #[derive(Clone, Debug)]
 pub struct PeerInfo<B: BlockT> {
@@ -433,72 +417,22 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	}
 
 	fn handle_response(&mut self, who: NodeIndex, response: &message::BlockResponse<B>) -> Option<message::BlockRequest<B>> {
-		let request = if let Some(ref mut peer) = self.context_data.peers.get_mut(&who) {
-			match (peer.block_request.take(), peer.justification_request.take()) {
-				(Some(block_request), Some(justification_request)) => {
-					if block_request.id == response.id {
-						peer.block_request_timestamp = None;
-						peer.justification_request = Some(justification_request);
-						block_request
-					} else if justification_request.id == response.id {
-						peer.justification_request_timestamp = None;
-						peer.block_request = Some(block_request);
-						justification_request
-					} else {
-						peer.justification_request_timestamp = None;
-						peer.block_request_timestamp = None;
-						trace!(target: "sync", "Ignoring mismatched response packet from {} (expected {} or {} got {})",
-							who,
-							block_request.id,
-							justification_request.id,
-							response.id,
-						);
-						return None;
-					}
-				},
-				(Some(block_request), None) => {
-					if block_request.id == response.id {
-						peer.block_request_timestamp = None;
-						block_request
-					} else {
-						peer.block_request_timestamp = None;
-						trace!(target: "sync", "Ignoring mismatched response packet from {} (expected {} got {})",
-							who,
-							block_request.id,
-							response.id,
-						);
-						return None;
-					}
-				},
-				(None, Some(justification_request)) => {
-					if justification_request.id == response.id {
-						peer.justification_request_timestamp = None;
-						justification_request
-					} else {
-						peer.justification_request_timestamp = None;
-						trace!(target: "sync", "Ignoring mismatched response packet from {} (expected {} got {})",
-							who,
-							justification_request.id,
-							response.id,
-						);
-						return None;
-					}
-				},
-				(None, None) => {
-					let _ = self
-						.network_chan
-						.send(NetworkMsg::ReportPeer(who, Severity::Bad("Unexpected response packet received from peer".to_string())));
-					return None;
-				},
+		if let Some(ref mut peer) = self.context_data.peers.get_mut(&who) {
+			if let Some(_) = peer.obsolete_requests.remove(&response.id) {
+				trace!(target: "sync", "Ignoring obsolete block response packet from {} ({})", who, response.id,);
+				return None;
 			}
-		} else {
+			// Clear the request. If the response is invalid peer will be disconnected anyway.
+			let request = peer.block_request.take();
+			if request.as_ref().map_or(false, |(_, r)| r.id == response.id) {
+				return request.map(|(_, r)| r)
+			}
+			trace!(target: "sync", "Unexpected response packet from {} ({})", who, response.id,);
 			let _ = self
 				.network_chan
-				.send(NetworkMsg::ReportPeer(who, Severity::Bad("Unexpected packet received from peer".to_string())));
-			return None;
-		};
-
-		Some(request)
+				.send(NetworkMsg::ReportPeer(who, Severity::Bad("Unexpected response packet received from peer".to_string())));
+		}
+		None
 	}
 
 	/// Returns protocol status
@@ -726,18 +660,19 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		let tick = time::Instant::now();
 		let mut aborting = Vec::new();
 		{
-			for (who, timestamp) in self
-				.context_data
-				.peers
-				.iter()
-				.filter_map(|(id, peer)| peer.min_request_timestamp().map(|r| (id, r)))
-				.chain(self.handshaking_peers.iter())
-			{
-				if (tick - *timestamp).as_secs() > REQUEST_TIMEOUT_SEC {
-					trace!(target: "sync", "Timeout {}", who);
+			for (who, peer) in self.context_data.peers.iter() {
+				if peer.block_request.as_ref().map_or(false, |(t, _)| (tick - *t).as_secs() > REQUEST_TIMEOUT_SEC) {
+					trace!(target: "sync", "Reqeust timeout {}", who);
+					aborting.push(*who);
+				} else if peer.obsolete_requests.values().any(|t| (tick - *t).as_secs() > REQUEST_TIMEOUT_SEC) {
+					trace!(target: "sync", "Obsolete timeout {}", who);
 					aborting.push(*who);
 				}
 			}
+			for (who, _) in self.handshaking_peers.iter().filter(|(_, t)| (tick - **t).as_secs() > REQUEST_TIMEOUT_SEC) {
+				trace!(target: "sync", "Handshake timeout {}", who);
+				aborting.push(*who);
+			}
 		}
 
 		self.specialization
@@ -819,12 +754,10 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 				best_hash: status.best_hash,
 				best_number: status.best_number,
 				block_request: None,
-				block_request_timestamp: None,
-				justification_request: None,
-				justification_request_timestamp: None,
 				known_extrinsics: HashSet::new(),
 				known_blocks: HashSet::new(),
 				next_request_id: 0,
+				obsolete_requests: HashMap::new(),
 			};
 			self.context_data.peers.insert(who.clone(), peer);
 			self.handshaking_peers.remove(&who);
@@ -1182,14 +1115,11 @@ fn send_message<B: BlockT, H: ExHashT>(
 			if let Some(ref mut peer) = peers.get_mut(&who) {
 				r.id = peer.next_request_id;
 				peer.next_request_id = peer.next_request_id + 1;
-
-				if r.fields == message::BlockAttributes::JUSTIFICATION {
-					peer.justification_request = Some(r.clone());
-					peer.justification_request_timestamp = Some(time::Instant::now());
-				} else {
-					peer.block_request = Some(r.clone());
-					peer.block_request_timestamp = Some(time::Instant::now());
+				if let Some((timestamp, request)) = peer.block_request.take() {
+					trace!(target: "sync", "Request {} for {} is now obsolete.", request.id, who);
+					peer.obsolete_requests.insert(request.id, timestamp);
 				}
+				peer.block_request = Some((time::Instant::now(), r.clone()));
 			}
 		}
 		_ => (),

substrate/core/network/src/test/block_import.rs

@@ -129,8 +129,9 @@ fn process_import_result_works() {
 
 	let link = TestLink::new();
 	assert_eq!(process_import_result::<Block>(&link, Err(BlockImportError::IncompleteHeader(Some(0)))), 0);
-	assert_eq!(link.total(), 1);
+	assert_eq!(link.total(), 2);
 	assert_eq!(link.disconnects.get(), 1);
+	assert_eq!(link.restarts.get(), 1);
 
 	let link = TestLink::new();
 	assert_eq!(process_import_result::<Block>(&link, Err(BlockImportError::UnknownParent)), 0);
@@ -141,6 +142,12 @@ fn process_import_result_works() {
 	assert_eq!(process_import_result::<Block>(&link, Err(BlockImportError::Error)), 0);
 	assert_eq!(link.total(), 1);
 	assert_eq!(link.restarts.get(), 1);
+
+	let link = TestLink::new();
+	assert_eq!(process_import_result::<Block>(&link, Err(BlockImportError::VerificationFailed(Some(0), String::new()))), 0);
+	assert_eq!(link.total(), 2);
+	assert_eq!(link.restarts.get(), 1);
+	assert_eq!(link.disconnects.get(), 1);
 }
 
 #[test]