Make BlockImport and Verifier async (#8472)

* Make grandpa work

* Introduce `SharedData`

* Add test and fix bugs

* Switch to `SharedData`

* Make grandpa tests working

* More Babe work

* Make it async

* Fix fix

* Use `async_trait` in sc-consensus-slots

This makes the code a little bit easier to read and also expresses that
there can always only be one call at a time to `on_slot`.

* Make grandpa tests compile

* More Babe tests work

* Fix network test

* Start fixing service test

* Finish service-test

* Fix sc-consensus-aura

* Fix fix fix

* More fixes

* Make everything compile *yeah*

* Fix build when we have Rust 1.51

* Update client/consensus/common/src/shared_data.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/common/src/shared_data.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/common/src/shared_data.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/common/src/shared_data.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/common/src/shared_data.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/babe/src/tests.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Update client/consensus/babe/src/tests.rs

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Fix warning

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

substrate/client/consensus/babe/src/aux_schema.rs

@@ -18,8 +18,6 @@
 
 //! Schema for BABE epoch changes in the aux-db.
 
-use std::sync::Arc;
-use parking_lot::Mutex;
 use log::info;
 use codec::{Decode, Encode};
 
@@ -79,18 +77,19 @@ pub fn load_epoch_changes<Block: BlockT, B: AuxStore>(
 		},
 	};
 
-	let epoch_changes = Arc::new(Mutex::new(maybe_epoch_changes.unwrap_or_else(|| {
-		info!(target: "babe",
-			  "👶 Creating empty BABE epoch changes on what appears to be first startup."
+	let epoch_changes = SharedEpochChanges::<Block, Epoch>::new(maybe_epoch_changes.unwrap_or_else(|| {
+		info!(
+			target: "babe",
+			"👶 Creating empty BABE epoch changes on what appears to be first startup.",
 		);
 		EpochChangesFor::<Block, Epoch>::default()
-	})));
+	}));
 
 	// rebalance the tree after deserialization. this isn't strictly necessary
 	// since the tree is now rebalanced on every update operation. but since the
 	// tree wasn't rebalanced initially it's useful to temporarily leave it here
 	// to avoid having to wait until an import for rebalancing.
-	epoch_changes.lock().rebalance();
+	epoch_changes.shared_data().rebalance();
 
 	Ok(epoch_changes)
 }
@@ -189,7 +188,7 @@ mod test {
 		).unwrap();
 
 		assert!(
-			epoch_changes.lock()
+			epoch_changes.shared_data()
 				.tree()
 				.iter()
 				.map(|(_, _, epoch)| epoch.clone())
@@ -201,7 +200,7 @@ mod test {
 		); // PersistedEpochHeader does not implement Debug, so we use assert! directly.
 
 		write_epoch_changes::<TestBlock, _, _>(
-			&epoch_changes.lock(),
+			&epoch_changes.shared_data(),
 			|values| {
 				client.insert_aux(values, &[]).unwrap();
 			},

substrate/client/consensus/babe/src/lib.rs

@@ -76,8 +76,8 @@ pub use sp_consensus_babe::{
 pub use sp_consensus::SyncOracle;
 pub use sc_consensus_slots::SlotProportion;
 use std::{
-	collections::HashMap, sync::Arc, u64, pin::Pin, time::{Instant, Duration},
-	any::Any, borrow::Cow, convert::TryInto,
+	collections::HashMap, sync::Arc, u64, pin::Pin, borrow::Cow, convert::TryInto,
+	time::{Duration, Instant},
 };
 use sp_consensus::{ImportResult, CanAuthorWith, import_queue::BoxJustificationImport};
 use sp_core::crypto::Public;
@@ -502,7 +502,7 @@ async fn answer_requests<B: BlockT, C>(
 		match request {
 			BabeRequest::EpochForChild(parent_hash, parent_number, slot_number, response) => {
 				let lookup = || {
-					let epoch_changes =	epoch_changes.lock();
+					let epoch_changes = epoch_changes.shared_data();
 					let epoch_descriptor = epoch_changes.epoch_descriptor_for_child_of(
 						descendent_query(&*client),
 						&parent_hash,
@@ -656,7 +656,7 @@ where
 		parent: &B::Header,
 		slot: Slot,
 	) -> Result<Self::EpochData, ConsensusError> {
-		self.epoch_changes.lock().epoch_descriptor_for_child_of(
+		self.epoch_changes.shared_data().epoch_descriptor_for_child_of(
 			descendent_query(&*self.client),
 			&parent.hash(),
 			parent.number().clone(),
@@ -667,7 +667,8 @@ where
 	}
 
 	fn authorities_len(&self, epoch_descriptor: &Self::EpochData) -> Option<usize> {
-		self.epoch_changes.lock()
+		self.epoch_changes
+			.shared_data()
 			.viable_epoch(&epoch_descriptor, |slot| Epoch::genesis(&self.config, slot))
 			.map(|epoch| epoch.as_ref().authorities.len())
 	}
@@ -681,7 +682,7 @@ where
 		debug!(target: "babe", "Attempting to claim slot {}", slot);
 		let s = authorship::claim_slot(
 			slot,
-			self.epoch_changes.lock().viable_epoch(
+			self.epoch_changes.shared_data().viable_epoch(
 				&epoch_descriptor,
 				|slot| Epoch::genesis(&self.config, slot)
 			)?.as_ref(),
@@ -768,7 +769,7 @@ where
 			import_block.storage_changes = Some(storage_changes);
 			import_block.intermediates.insert(
 				Cow::from(INTERMEDIATE_KEY),
-				Box::new(BabeIntermediate::<B> { epoch_descriptor }) as Box<dyn Any>,
+				Box::new(BabeIntermediate::<B> { epoch_descriptor }) as Box<_>,
 			);
 
 			Ok(import_block)
@@ -1083,6 +1084,7 @@ where
 	}
 }
 
+#[async_trait::async_trait]
 impl<Block, Client, SelectChain, CAW> Verifier<Block>
 	for BabeVerifier<Block, Client, SelectChain, CAW>
 where
@@ -1093,7 +1095,7 @@ where
 	SelectChain: sp_consensus::SelectChain<Block>,
 	CAW: CanAuthorWith<Block> + Send + Sync,
 {
-	fn verify(
+	async fn verify(
 		&mut self,
 		origin: BlockOrigin,
 		header: Block::Header,
@@ -1125,7 +1127,7 @@ where
 			.map_err(Error::<Block>::FetchParentHeader)?;
 
 		let pre_digest = find_pre_digest::<Block>(&header)?;
-		let epoch_changes = self.epoch_changes.lock();
+		let epoch_changes = self.epoch_changes.shared_data();
 		let epoch_descriptor = epoch_changes.epoch_descriptor_for_child_of(
 			descendent_query(&*self.client),
 			&parent_hash,
@@ -1189,7 +1191,8 @@ where
 					self.telemetry;
 					CONSENSUS_TRACE;
 					"babe.checked_and_importing";
-					"pre_header" => ?pre_header);
+					"pre_header" => ?pre_header,
+				);
 
 				let mut import_block = BlockImportParams::new(origin, pre_header);
 				import_block.post_digests.push(verified_info.seal);
@@ -1197,7 +1200,7 @@ where
 				import_block.justifications = justifications;
 				import_block.intermediates.insert(
 					Cow::from(INTERMEDIATE_KEY),
-					Box::new(BabeIntermediate::<Block> { epoch_descriptor }) as Box<dyn Any>,
+					Box::new(BabeIntermediate::<Block> { epoch_descriptor }) as Box<_>,
 				);
 				import_block.post_hash = Some(hash);
 
@@ -1275,6 +1278,7 @@ impl<Block: BlockT, Client, I> BabeBlockImport<Block, Client, I> {
 	}
 }
 
+#[async_trait::async_trait]
 impl<Block, Client, Inner> BlockImport<Block> for BabeBlockImport<Block, Client, Inner> where
 	Block: BlockT,
 	Inner: BlockImport<Block, Transaction = sp_api::TransactionFor<Client, Block>> + Send + Sync,
@@ -1286,7 +1290,7 @@ impl<Block, Client, Inner> BlockImport<Block> for BabeBlockImport<Block, Client,
 	type Error = ConsensusError;
 	type Transaction = sp_api::TransactionFor<Client, Block>;
 
-	fn import_block(
+	async fn import_block(
 		&mut self,
 		mut block: BlockImportParams<Block, Self::Transaction>,
 		new_cache: HashMap<CacheKeyId, Vec<u8>>,
@@ -1328,202 +1332,209 @@ impl<Block, Client, Inner> BlockImport<Block> for BabeBlockImport<Block, Client,
 			);
 		}
 
-		let mut epoch_changes = self.epoch_changes.lock();
-
-		// check if there's any epoch change expected to happen at this slot.
-		// `epoch` is the epoch to verify the block under, and `first_in_epoch` is true
-		// if this is the first block in its chain for that epoch.
-		//
-		// also provides the total weight of the chain, including the imported block.
-		let (epoch_descriptor, first_in_epoch, parent_weight) = {
-			let parent_weight = if *parent_header.number() == Zero::zero() {
-				0
-			} else {
-				aux_schema::load_block_weight(&*self.client, parent_hash)
-					.map_err(|e| ConsensusError::ClientImport(e.to_string()))?
-					.ok_or_else(|| ConsensusError::ClientImport(
-						babe_err(Error::<Block>::ParentBlockNoAssociatedWeight(hash)).into()
-					))?
-			};
-
-			let intermediate = block.take_intermediate::<BabeIntermediate<Block>>(
-				INTERMEDIATE_KEY
-			)?;
-
-			let epoch_descriptor = intermediate.epoch_descriptor;
-			let first_in_epoch = parent_slot < epoch_descriptor.start_slot();
-			(epoch_descriptor, first_in_epoch, parent_weight)
-		};
-
-		let total_weight = parent_weight + pre_digest.added_weight();
-
-		// search for this all the time so we can reject unexpected announcements.
-		let next_epoch_digest = find_next_epoch_digest::<Block>(&block.header)
-			.map_err(|e| ConsensusError::ClientImport(e.to_string()))?;
-		let next_config_digest = find_next_config_digest::<Block>(&block.header)
-			.map_err(|e| ConsensusError::ClientImport(e.to_string()))?;
-
-		match (first_in_epoch, next_epoch_digest.is_some(), next_config_digest.is_some()) {
-			(true, true, _) => {},
-			(false, false, false) => {},
-			(false, false, true) => {
-				return Err(
-					ConsensusError::ClientImport(
-						babe_err(Error::<Block>::UnexpectedConfigChange).into(),
-					)
-				)
-			},
-			(true, false, _) => {
-				return Err(
-					ConsensusError::ClientImport(
-						babe_err(Error::<Block>::ExpectedEpochChange(hash, slot)).into(),
-					)
-				)
-			},
-			(false, true, _) => {
-				return Err(
-					ConsensusError::ClientImport(
-						babe_err(Error::<Block>::UnexpectedEpochChange).into(),
-					)
-				)
-			},
-		}
-
 		// if there's a pending epoch we'll save the previous epoch changes here
 		// this way we can revert it if there's any error
 		let mut old_epoch_changes = None;
 
-		let info = self.client.info();
+		// Use an extra scope to make the compiler happy, because otherwise he complains about the
+		// mutex, even if we dropped it...
+		let mut epoch_changes = {
+			let mut epoch_changes = self.epoch_changes.shared_data_locked();
 
-		if let Some(next_epoch_descriptor) = next_epoch_digest {
-			old_epoch_changes = Some(epoch_changes.clone());
+			// check if there's any epoch change expected to happen at this slot.
+			// `epoch` is the epoch to verify the block under, and `first_in_epoch` is true
+			// if this is the first block in its chain for that epoch.
+			//
+			// also provides the total weight of the chain, including the imported block.
+			let (epoch_descriptor, first_in_epoch, parent_weight) = {
+				let parent_weight = if *parent_header.number() == Zero::zero() {
+					0
+				} else {
+					aux_schema::load_block_weight(&*self.client, parent_hash)
+						.map_err(|e| ConsensusError::ClientImport(e.to_string()))?
+					.ok_or_else(|| ConsensusError::ClientImport(
+						babe_err(Error::<Block>::ParentBlockNoAssociatedWeight(hash)).into()
+					))?
+				};
 
-			let viable_epoch = epoch_changes.viable_epoch(
-				&epoch_descriptor,
-				|slot| Epoch::genesis(&self.config, slot)
-			).ok_or_else(|| {
-				ConsensusError::ClientImport(Error::<Block>::FetchEpoch(parent_hash).into())
-			})?;
-
-			let epoch_config = next_config_digest.map(Into::into).unwrap_or_else(
-				|| viable_epoch.as_ref().config.clone()
-			);
-
-			// restrict info logging during initial sync to avoid spam
-			let log_level = if block.origin == BlockOrigin::NetworkInitialSync {
-				log::Level::Debug
-			} else {
-				log::Level::Info
-			};
-
-			log!(target: "babe",
-				log_level,
-				"👶 New epoch {} launching at block {} (block slot {} >= start slot {}).",
-				viable_epoch.as_ref().epoch_index,
-				hash,
-				slot,
-				viable_epoch.as_ref().start_slot,
-			);
-
-			let next_epoch = viable_epoch.increment((next_epoch_descriptor, epoch_config));
-
-			log!(target: "babe",
-				log_level,
-				"👶 Next epoch starts at slot {}",
-				next_epoch.as_ref().start_slot,
-			);
-
-			// prune the tree of epochs not part of the finalized chain or
-			// that are not live anymore, and then track the given epoch change
-			// in the tree.
-			// NOTE: it is important that these operations are done in this
-			// order, otherwise if pruning after import the `is_descendent_of`
-			// used by pruning may not know about the block that is being
-			// imported.
-			let prune_and_import = || {
-				prune_finalized(
-					self.client.clone(),
-					&mut epoch_changes,
+				let intermediate = block.take_intermediate::<BabeIntermediate<Block>>(
+					INTERMEDIATE_KEY
 				)?;
 
-				epoch_changes.import(
-					descendent_query(&*self.client),
-					hash,
-					number,
-					*block.header.parent_hash(),
-					next_epoch,
-				).map_err(|e| ConsensusError::ClientImport(format!("{:?}", e)))?;
-
-				Ok(())
+				let epoch_descriptor = intermediate.epoch_descriptor;
+				let first_in_epoch = parent_slot < epoch_descriptor.start_slot();
+				(epoch_descriptor, first_in_epoch, parent_weight)
 			};
 
-			if let Err(e) = prune_and_import() {
-				debug!(target: "babe", "Failed to launch next epoch: {:?}", e);
-				*epoch_changes = old_epoch_changes.expect("set `Some` above and not taken; qed");
-				return Err(e);
+			let total_weight = parent_weight + pre_digest.added_weight();
+
+			// search for this all the time so we can reject unexpected announcements.
+			let next_epoch_digest = find_next_epoch_digest::<Block>(&block.header)
+				.map_err(|e| ConsensusError::ClientImport(e.to_string()))?;
+			let next_config_digest = find_next_config_digest::<Block>(&block.header)
+				.map_err(|e| ConsensusError::ClientImport(e.to_string()))?;
+
+			match (first_in_epoch, next_epoch_digest.is_some(), next_config_digest.is_some()) {
+				(true, true, _) => {},
+				(false, false, false) => {},
+				(false, false, true) => {
+					return Err(
+						ConsensusError::ClientImport(
+							babe_err(Error::<Block>::UnexpectedConfigChange).into(),
+						)
+					)
+				},
+				(true, false, _) => {
+					return Err(
+						ConsensusError::ClientImport(
+							babe_err(Error::<Block>::ExpectedEpochChange(hash, slot)).into(),
+						)
+					)
+				},
+				(false, true, _) => {
+					return Err(
+						ConsensusError::ClientImport(
+							babe_err(Error::<Block>::UnexpectedEpochChange).into(),
+						)
+					)
+				},
 			}
 
-			crate::aux_schema::write_epoch_changes::<Block, _, _>(
-				&*epoch_changes,
-				|insert| block.auxiliary.extend(
-					insert.iter().map(|(k, v)| (k.to_vec(), Some(v.to_vec())))
-				)
+			let info = self.client.info();
+
+			if let Some(next_epoch_descriptor) = next_epoch_digest {
+				old_epoch_changes = Some((*epoch_changes).clone());
+
+				let viable_epoch = epoch_changes.viable_epoch(
+					&epoch_descriptor,
+					|slot| Epoch::genesis(&self.config, slot)
+				).ok_or_else(|| {
+					ConsensusError::ClientImport(Error::<Block>::FetchEpoch(parent_hash).into())
+				})?;
+
+				let epoch_config = next_config_digest.map(Into::into).unwrap_or_else(
+					|| viable_epoch.as_ref().config.clone()
+				);
+
+				// restrict info logging during initial sync to avoid spam
+				let log_level = if block.origin == BlockOrigin::NetworkInitialSync {
+					log::Level::Debug
+				} else {
+					log::Level::Info
+				};
+
+				log!(target: "babe",
+					 log_level,
+					 "👶 New epoch {} launching at block {} (block slot {} >= start slot {}).",
+					 viable_epoch.as_ref().epoch_index,
+					 hash,
+					 slot,
+					 viable_epoch.as_ref().start_slot,
+				);
+
+				let next_epoch = viable_epoch.increment((next_epoch_descriptor, epoch_config));
+
+				log!(target: "babe",
+					 log_level,
+					 "👶 Next epoch starts at slot {}",
+					 next_epoch.as_ref().start_slot,
+				);
+
+				// prune the tree of epochs not part of the finalized chain or
+				// that are not live anymore, and then track the given epoch change
+				// in the tree.
+				// NOTE: it is important that these operations are done in this
+				// order, otherwise if pruning after import the `is_descendent_of`
+				// used by pruning may not know about the block that is being
+				// imported.
+				let prune_and_import = || {
+					prune_finalized(
+						self.client.clone(),
+						&mut epoch_changes,
+					)?;
+
+					epoch_changes.import(
+						descendent_query(&*self.client),
+						hash,
+						number,
+						*block.header.parent_hash(),
+						next_epoch,
+					).map_err(|e| ConsensusError::ClientImport(format!("{:?}", e)))?;
+
+					Ok(())
+				};
+
+				if let Err(e) = prune_and_import() {
+					debug!(target: "babe", "Failed to launch next epoch: {:?}", e);
+					*epoch_changes = old_epoch_changes.expect("set `Some` above and not taken; qed");
+					return Err(e);
+				}
+
+				crate::aux_schema::write_epoch_changes::<Block, _, _>(
+					&*epoch_changes,
+					|insert| block.auxiliary.extend(
+						insert.iter().map(|(k, v)| (k.to_vec(), Some(v.to_vec())))
+					)
+				);
+			}
+
+			aux_schema::write_block_weight(
+				hash,
+				total_weight,
+				|values| block.auxiliary.extend(
+					values.iter().map(|(k, v)| (k.to_vec(), Some(v.to_vec())))
+				),
 			);
-		}
 
-		aux_schema::write_block_weight(
-			hash,
-			total_weight,
-			|values| block.auxiliary.extend(
-				values.iter().map(|(k, v)| (k.to_vec(), Some(v.to_vec())))
-			),
-		);
+			// The fork choice rule is that we pick the heaviest chain (i.e.
+			// more primary blocks), if there's a tie we go with the longest
+			// chain.
+			block.fork_choice = {
+				let (last_best, last_best_number) = (info.best_hash, info.best_number);
 
-		// The fork choice rule is that we pick the heaviest chain (i.e.
-		// more primary blocks), if there's a tie we go with the longest
-		// chain.
-		block.fork_choice = {
-			let (last_best, last_best_number) = (info.best_hash, info.best_number);
-
-			let last_best_weight = if &last_best == block.header.parent_hash() {
-				// the parent=genesis case is already covered for loading parent weight,
-				// so we don't need to cover again here.
-				parent_weight
-			} else {
-				aux_schema::load_block_weight(&*self.client, last_best)
-					.map_err(|e| ConsensusError::ChainLookup(format!("{:?}", e)))?
+				let last_best_weight = if &last_best == block.header.parent_hash() {
+					// the parent=genesis case is already covered for loading parent weight,
+					// so we don't need to cover again here.
+					parent_weight
+				} else {
+					aux_schema::load_block_weight(&*self.client, last_best)
+						.map_err(|e| ConsensusError::ChainLookup(format!("{:?}", e)))?
 					.ok_or_else(
 						|| ConsensusError::ChainLookup("No block weight for parent header.".to_string())
 					)?
+				};
+
+				Some(ForkChoiceStrategy::Custom(if total_weight > last_best_weight {
+					true
+				} else if total_weight == last_best_weight {
+					number > last_best_number
+				} else {
+					false
+				}))
 			};
 
-			Some(ForkChoiceStrategy::Custom(if total_weight > last_best_weight {
-				true
-			} else if total_weight == last_best_weight {
-				number > last_best_number
-			} else {
-				false
-			}))
+			// Release the mutex, but it stays locked
+			epoch_changes.release_mutex()
 		};
 
-		let import_result = self.inner.import_block(block, new_cache);
+		let import_result = self.inner.import_block(block, new_cache).await;
 
 		// revert to the original epoch changes in case there's an error
 		// importing the block
 		if import_result.is_err() {
 			if let Some(old_epoch_changes) = old_epoch_changes {
-				*epoch_changes = old_epoch_changes;
+				*epoch_changes.upgrade() = old_epoch_changes;
 			}
 		}
 
 		import_result.map_err(Into::into)
 	}
 
-	fn check_block(
+	async fn check_block(
 		&mut self,
 		block: BlockCheckParams<Block>,
 	) -> Result<ImportResult, Self::Error> {
-		self.inner.check_block(block).map_err(Into::into)
+		self.inner.check_block(block).await.map_err(Into::into)
 	}
 }
 
@@ -1583,7 +1594,7 @@ pub fn block_import<Client, Block: BlockT, I>(
 	// startup rather than waiting until importing the next epoch change block.
 	prune_finalized(
 		client.clone(),
-		&mut epoch_changes.lock(),
+		&mut epoch_changes.shared_data(),
 	)?;
 
 	let import = BabeBlockImport::new(

substrate/client/consensus/babe/src/tests.rs

@@ -47,6 +47,7 @@ use rand_chacha::{
 };
 use sc_keystore::LocalKeystore;
 use sp_application_crypto::key_types::BABE;
+use futures::executor::block_on;
 
 type Item = DigestItem<Hash>;
 
@@ -67,6 +68,9 @@ enum Stage {
 
 type Mutator = Arc<dyn Fn(&mut TestHeader, Stage) + Send + Sync>;
 
+type BabeBlockImport =
+	PanickingBlockImport<crate::BabeBlockImport<TestBlock, TestClient, Arc<TestClient>>>;
+
 #[derive(Clone)]
 struct DummyFactory {
 	client: Arc<TestClient>,
@@ -134,7 +138,7 @@ impl DummyProposer {
 
 		// figure out if we should add a consensus digest, since the test runtime
 		// doesn't.
-		let epoch_changes = self.factory.epoch_changes.lock();
+		let epoch_changes = self.factory.epoch_changes.shared_data();
 		let epoch = epoch_changes.epoch_data_for_child_of(
 			descendent_query(&*self.factory.client),
 			&self.parent_hash,
@@ -188,30 +192,37 @@ thread_local! {
 }
 
 #[derive(Clone)]
-struct PanickingBlockImport<B>(B);
+pub struct PanickingBlockImport<B>(B);
 
-impl<B: BlockImport<TestBlock>> BlockImport<TestBlock> for PanickingBlockImport<B> {
+#[async_trait::async_trait]
+impl<B: BlockImport<TestBlock>> BlockImport<TestBlock> for PanickingBlockImport<B>
+	where
+		B::Transaction: Send,
+		B: Send,
+{
 	type Error = B::Error;
 	type Transaction = B::Transaction;
 
-	fn import_block(
+	async fn import_block(
 		&mut self,
 		block: BlockImportParams<TestBlock, Self::Transaction>,
 		new_cache: HashMap<CacheKeyId, Vec<u8>>,
 	) -> Result<ImportResult, Self::Error> {
-		Ok(self.0.import_block(block, new_cache).expect("importing block failed"))
+		Ok(self.0.import_block(block, new_cache).await.expect("importing block failed"))
 	}
 
-	fn check_block(
+	async fn check_block(
 		&mut self,
 		block: BlockCheckParams<TestBlock>,
 	) -> Result<ImportResult, Self::Error> {
-		Ok(self.0.check_block(block).expect("checking block failed"))
+		Ok(self.0.check_block(block).await.expect("checking block failed"))
 	}
 }
 
+type BabePeer = Peer<Option<PeerData>, BabeBlockImport>;
+
 pub struct BabeTestNet {
-	peers: Vec<Peer<Option<PeerData>>>,
+	peers: Vec<BabePeer>,
 }
 
 type TestHeader = <TestBlock as BlockT>::Header;
@@ -227,11 +238,12 @@ pub struct TestVerifier {
 	mutator: Mutator,
 }
 
+#[async_trait::async_trait]
 impl Verifier<TestBlock> for TestVerifier {
 	/// Verify the given data and return the BlockImportParams and an optional
 	/// new set of validators to import. If not, err with an Error-Message
 	/// presented to the User in the logs.
-	fn verify(
+	async fn verify(
 		&mut self,
 		origin: BlockOrigin,
 		mut header: TestHeader,
@@ -240,7 +252,7 @@ impl Verifier<TestBlock> for TestVerifier {
 	) -> Result<(BlockImportParams<TestBlock, ()>, Option<Vec<(CacheKeyId, Vec<u8>)>>), String> {
 		// apply post-sealing mutations (i.e. stripping seal, if desired).
 		(self.mutator)(&mut header, Stage::PostSeal);
-		self.inner.verify(origin, header, justifications, body)
+		self.inner.verify(dbg!(origin), header, justifications, body).await
 	}
 }
 
@@ -255,6 +267,7 @@ pub struct PeerData {
 impl TestNetFactory for BabeTestNet {
 	type Verifier = TestVerifier;
 	type PeerData = Option<PeerData>;
+	type BlockImport = BabeBlockImport;
 
 	/// Create new test network with peers and given config.
 	fn from_config(_config: &ProtocolConfig) -> Self {
@@ -264,9 +277,9 @@ impl TestNetFactory for BabeTestNet {
 		}
 	}
 
-	fn make_block_import<Transaction>(&self, client: PeersClient)
+	fn make_block_import(&self, client: PeersClient)
 		-> (
-			BlockImportAdapter<Transaction>,
+			BlockImportAdapter<Self::BlockImport>,
 			Option<BoxJustificationImport<Block>>,
 			Option<PeerData>,
 		)
@@ -287,7 +300,7 @@ impl TestNetFactory for BabeTestNet {
 			Some(Box::new(block_import.clone()) as BoxBlockImport<_, _>)
 		);
 		(
-			BlockImportAdapter::new_full(block_import),
+			BlockImportAdapter::new(block_import),
 			None,
 			Some(PeerData { link, inherent_data_providers, block_import: data_block_import }),
 		)
@@ -326,17 +339,17 @@ impl TestNetFactory for BabeTestNet {
 		}
 	}
 
-	fn peer(&mut self, i: usize) -> &mut Peer<Self::PeerData> {
+	fn peer(&mut self, i: usize) -> &mut BabePeer {
 		trace!(target: "babe", "Retrieving a peer");
 		&mut self.peers[i]
 	}
 
-	fn peers(&self) -> &Vec<Peer<Self::PeerData>> {
+	fn peers(&self) -> &Vec<BabePeer> {
 		trace!(target: "babe", "Retrieving peers");
 		&self.peers
 	}
 
-	fn mut_peers<F: FnOnce(&mut Vec<Peer<Self::PeerData>>)>(
+	fn mut_peers<F: FnOnce(&mut Vec<BabePeer>)>(
 		&mut self,
 		closure: F,
 	) {
@@ -436,7 +449,7 @@ fn run_one_test(
 			telemetry: None,
 		}).expect("Starts babe"));
 	}
-	futures::executor::block_on(future::select(
+	block_on(future::select(
 		futures::future::poll_fn(move |cx| {
 			let mut net = net.lock();
 			net.poll(cx);
@@ -567,7 +580,7 @@ fn can_author_block() {
 }
 
 // Propose and import a new BABE block on top of the given parent.
-fn propose_and_import_block<Transaction>(
+fn propose_and_import_block<Transaction: Send + 'static>(
 	parent: &TestHeader,
 	slot: Option<Slot>,
 	proposer_factory: &mut DummyFactory,
@@ -595,7 +608,7 @@ fn propose_and_import_block<Transaction>(
 
 	let mut block = futures::executor::block_on(proposer.propose_with(pre_digest)).unwrap().block;
 
-	let epoch_descriptor = proposer_factory.epoch_changes.lock().epoch_descriptor_for_child_of(
+	let epoch_descriptor = proposer_factory.epoch_changes.shared_data().epoch_descriptor_for_child_of(
 		descendent_query(&*proposer_factory.client),
 		&parent_hash,
 		*parent.number(),
@@ -623,10 +636,10 @@ fn propose_and_import_block<Transaction>(
 	import.body = Some(block.extrinsics);
 	import.intermediates.insert(
 		Cow::from(INTERMEDIATE_KEY),
-		Box::new(BabeIntermediate::<TestBlock> { epoch_descriptor }) as Box<dyn Any>,
+		Box::new(BabeIntermediate::<TestBlock> { epoch_descriptor }) as Box<_>,
 	);
 	import.fork_choice = Some(ForkChoiceStrategy::LongestChain);
-	let import_result = block_import.import_block(import, Default::default()).unwrap();
+	let import_result = block_on(block_import.import_block(import, Default::default())).unwrap();
 
 	match import_result {
 		ImportResult::Imported(_) => {},
@@ -664,7 +677,7 @@ fn importing_block_one_sets_genesis_epoch() {
 
 	let genesis_epoch = Epoch::genesis(&data.link.config, 999.into());
 
-	let epoch_changes = data.link.epoch_changes.lock();
+	let epoch_changes = data.link.epoch_changes.shared_data();
 	let epoch_for_second_block = epoch_changes.epoch_data_for_child_of(
 		descendent_query(&*client),
 		&block_hash,
@@ -739,13 +752,13 @@ fn importing_epoch_change_block_prunes_tree() {
 
 	// We should be tracking a total of 9 epochs in the fork tree
 	assert_eq!(
-		epoch_changes.lock().tree().iter().count(),
+		epoch_changes.shared_data().tree().iter().count(),
 		9,
 	);
 
 	// And only one root
 	assert_eq!(
-		epoch_changes.lock().tree().roots().count(),
+		epoch_changes.shared_data().tree().roots().count(),
 		1,
 	);
 
@@ -756,16 +769,16 @@ fn importing_epoch_change_block_prunes_tree() {
 
 	// at this point no hashes from the first fork must exist on the tree
 	assert!(
-		!epoch_changes.lock().tree().iter().map(|(h, _, _)| h).any(|h| fork_1.contains(h)),
+		!epoch_changes.shared_data().tree().iter().map(|(h, _, _)| h).any(|h| fork_1.contains(h)),
 	);
 
 	// but the epoch changes from the other forks must still exist
 	assert!(
-		epoch_changes.lock().tree().iter().map(|(h, _, _)| h).any(|h| fork_2.contains(h))
+		epoch_changes.shared_data().tree().iter().map(|(h, _, _)| h).any(|h| fork_2.contains(h))
 	);
 
 	assert!(
-		epoch_changes.lock().tree().iter().map(|(h, _, _)| h).any(|h| fork_3.contains(h)),
+		epoch_changes.shared_data().tree().iter().map(|(h, _, _)| h).any(|h| fork_3.contains(h)),
 	);
 
 	// finalizing block #25 from the canon chain should prune out the second fork
@@ -774,12 +787,12 @@ fn importing_epoch_change_block_prunes_tree() {
 
 	// at this point no hashes from the second fork must exist on the tree
 	assert!(
-		!epoch_changes.lock().tree().iter().map(|(h, _, _)| h).any(|h| fork_2.contains(h)),
+		!epoch_changes.shared_data().tree().iter().map(|(h, _, _)| h).any(|h| fork_2.contains(h)),
 	);
 
 	// while epoch changes from the last fork should still exist
 	assert!(
-		epoch_changes.lock().tree().iter().map(|(h, _, _)| h).any(|h| fork_3.contains(h)),
+		epoch_changes.shared_data().tree().iter().map(|(h, _, _)| h).any(|h| fork_3.contains(h)),
 	);
 }