Merge pull request #99 from paritytech/fix_known_blocks_are_ignored_in_headers_verification_success

Fixed panic when header is verifying for (Verifying | Stored) block

sync/src/synchronization_chain.rs

@@ -310,11 +310,12 @@ impl Chain {
 		self.verifying_headers.extend(headers.iter().map(|h| h.hash))
 	}
 
-	/// Remove headers from verifying queue
-	pub fn headers_verified(&mut self, headers: &[IndexedBlockHeader]) {
-		for header in headers {
-			self.verifying_headers.remove(&header.hash);
-		}
+	/// Remove headers from verifying queue.
+	///
+	/// Returns all headers that still have VerifyingHeader state (i.e. they are not Verifying || Stored).
+	pub fn headers_verified(&mut self, mut headers: Vec<IndexedBlockHeader>) -> Vec<IndexedBlockHeader> {
+		headers.retain(|header| self.verifying_headers.remove(&header.hash));
+		headers
 	}
 
 	/// Schedule blocks hashes for requesting
@@ -336,6 +337,9 @@ impl Chain {
 	/// chain, guarantees the header has already been pre-verified. The opposite isn't true -
 	/// if the header isn't in the chain, it could have been (in rare cases) pre-verified.
 	pub fn verify_block(&mut self, header: IndexedBlockHeader) -> bool {
+		// when we start verifying block, forget that we (possibly) verifying header of the block
+		self.verifying_headers.remove(&header.hash);
+
 		// insert header to the in-memory chain in case when it is not already there (non-headers-first sync)
 		self.hash_chain.push_back_at(VERIFYING_QUEUE, header.hash.clone());
 		self.headers_chain.insert(header)
@@ -362,7 +366,7 @@ impl Chain {
 		match block_origin {
 			storage::BlockOrigin::KnownBlock => {
 				// there should be no known blocks at this point
-				unreachable!();
+				unreachable!("Trying to re-insert known block: {}", block.hash().to_reversed_str());
 			},
 			// case 1: block has been added to the main branch
 			storage::BlockOrigin::CanonChain { .. } => {

sync/src/synchronization_client_core.rs

@@ -255,7 +255,7 @@ impl<T> ClientCore for SynchronizationClientCore<T> where T: TaskExecutor {
 
 	/// Try to queue synchronization of unknown blocks when blocks headers are received.
 	fn on_headers(&mut self, peer_index: PeerIndex, headers: Vec<IndexedBlockHeader>) -> Option<Vec<IndexedBlockHeader>> {
-		assert!(!	headers.is_empty(), "This must be checked in incoming connection");
+		assert!(!headers.is_empty(), "This is checked in incoming connection");
 
 		// update peers to select next tasks
 		self.peers_tasks.on_headers_received(peer_index);
@@ -421,6 +421,13 @@ impl<T> ClientCore for SynchronizationClientCore<T> where T: TaskExecutor {
 							self.chain.forget_block_leave_header(&block.header.hash);
 							// remember this block as unknown
 							if !self.orphaned_blocks_pool.contains_unknown_block(&block.header.hash) {
+								trace!(
+									target: "sync",
+									"Inserting unknown orphan block: {}. Block state: {:?}, parent state: {:?}",
+									block.header.hash.to_reversed_str(),
+									block_state,
+									parent_block_state,
+								);
 								self.orphaned_blocks_pool.insert_unknown_block(block);
 							}
 						}
@@ -455,9 +462,27 @@ impl<T> ClientCore for SynchronizationClientCore<T> where T: TaskExecutor {
 								entry.insert((blocks_to_verify_hashes.into_iter().collect(), Vec::new()));
 							}
 						}
+
+						trace!(
+							target: "sync",
+							"Scheduling verification of blocks: {}..{} First block state: {:?}, parent state: {:?}",
+							blocks_to_verify[0].hash().to_reversed_str(),
+							blocks_to_verify[blocks_to_verify.len() - 1].hash().to_reversed_str(),
+							block_state,
+							parent_block_state,
+						);
+
 						result = Some(blocks_to_verify);
 					},
 					BlockState::Requested | BlockState::Scheduled => {
+						trace!(
+							target: "sync",
+							"Inserting known orphan block: {}. Block state: {:?}, parent state: {:?}",
+							block.header.hash.to_reversed_str(),
+							block_state,
+							parent_block_state,
+						);
+
 						// remember peer as useful
 						self.peers_tasks.useful_peer(peer_index);
 						// remember as orphan block
@@ -1058,9 +1083,16 @@ impl<T> SynchronizationClientCore<T> where T: TaskExecutor {
 	}
 
 	fn on_headers_verification_success(&mut self, headers: Vec<IndexedBlockHeader>) {
-		self.chain.headers_verified(&headers);
-
-		self.chain.schedule_blocks_headers(headers);
+		let headers = self.chain.headers_verified(headers);
+		if !headers.is_empty() {
+			trace!(
+				target: "sync",
+				"Scheduling retrieval of headers: {}..{}",
+				headers[0].hash.to_reversed_str(),
+				headers[headers.len() - 1].hash.to_reversed_str(),
+			);
+			self.chain.schedule_blocks_headers(headers);
+		}
 
 		// switch to synchronization state
 		if !self.state.is_synchronizing() {
@@ -1076,7 +1108,7 @@ impl<T> SynchronizationClientCore<T> where T: TaskExecutor {
 	}
 
 	fn on_headers_verification_error(&mut self, peer: PeerIndex, error: String, hash: H256, headers: Vec<IndexedBlockHeader>) {
-		self.chain.headers_verified(&headers);
+		self.chain.headers_verified(headers);
 
 		if self.config.close_connection_on_bad_block {
 			self.peers.misbehaving(
@@ -1305,7 +1337,7 @@ pub mod tests {
 
 	use std::sync::Arc;
 	use parking_lot::{Mutex, RwLock};
-	use chain::{Block, Transaction};
+	use chain::{Block, Transaction, IndexedBlock};
 	use db::BlockChainDatabase;
 	use message::common::InventoryVector;
 	use message::{Services, types};
@@ -1314,7 +1346,7 @@ pub mod tests {
 	use primitives::hash::H256;
 	use verification::BackwardsCompatibleChainVerifier as ChainVerifier;
 	use inbound_connection::tests::DummyOutboundSyncConnection;
-	use synchronization_chain::Chain;
+	use synchronization_chain::{Chain, BlockState};
 	use synchronization_client::{SynchronizationClient, Client};
 	use synchronization_peers::PeersImpl;
 	use synchronization_executor::Task;
@@ -2459,4 +2491,59 @@ pub mod tests {
 		assert_eq!(data.lock().is_synchronizing, false);
 		assert_eq!(data.lock().best_blocks.len(), 3);
 	}
+
+	#[test]
+	fn known_blocks_are_ignored_in_headers_verification_success() {
+		let (_, sync, _) = create_sync(None, None);
+		let mut sync = sync.lock();
+
+		let block1: IndexedBlock = test_data::block_h1().into();
+		let block2: IndexedBlock = test_data::block_h2().into();
+		let header1 = block1.header.clone();
+		let header2 = block2.header.clone();
+		let hash1 = *block1.hash();
+		let hash2 = *block2.hash();
+
+		// WHEN
+		// we have orphaned [block2]
+		sync.orphaned_blocks_pool.insert_unknown_block(block2.clone());
+
+		// THEN:
+
+		// [header1] received => [header1] verification starts
+		sync.on_headers(0, vec![header1.clone()]);
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::VerifyingHeader);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Unknown);
+
+		// [header1] verification ends => [block1] is requested
+		sync.on_headers_verification_success(vec![header1.clone()]);
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Requested);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Unknown);
+
+		// [header2] received => [header2] verification starts
+		sync.on_headers(0, vec![header2.clone()]);
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Requested);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::VerifyingHeader);
+
+		// [block1] received => [block1, block2] verification starts
+		sync.on_block(0, block1.clone());
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Verifying);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Verifying); // pre-fix: VerifyingHeader
+
+		// [block1, block2] verification ends => [block1, block2] are inserted into DB
+		sync.on_block_verification_success(block1.clone());
+		sync.on_block_verification_success(block2.clone());
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Stored);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Stored); // pre-fix: VerifyingHeader
+
+		// [header2] verification ends => [block2] is requested
+		sync.on_headers_verification_success(vec![header2.clone()]);
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Stored);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Stored); // pre-fix: Requested
+
+		// [block2] received => [block2] verification starts
+		sync.on_block(0, block2.clone());
+		assert_eq!(sync.chain().block_state(&hash1), BlockState::Stored);
+		assert_eq!(sync.chain().block_state(&hash2), BlockState::Stored); // pre-fix: Verifying
+	}
 }

verification/src/chain_verifier.rs

@@ -53,7 +53,7 @@ impl BackwardsCompatibleChainVerifier {
 		match block_origin {
 			BlockOrigin::KnownBlock => {
 				// there should be no known blocks at this point
-				unreachable!();
+				unreachable!("Trying to re-verify known block: {}", block.hash().reversed());
 			},
 			BlockOrigin::CanonChain { block_number } => {
 				let tx_out_provider = CachedTransactionOutputProvider::new(self.store.as_store().as_transaction_output_provider());