Split execute_and_import_block in client into two functions (#1455)

* type error

* borrow problem

* compiles but run error

* tests pass ready for review

* tests pass ready for review

* fix comments by rob

* removed clone

* style fix

* updating

* resolved conflicts

* fixed comments

substrate/core/client/src/client.rs

@@ -43,7 +43,7 @@ use state_machine::{
 	DBValue, Backend as StateBackend, CodeExecutor, ChangesTrieAnchorBlockId,
 	ExecutionStrategy, ExecutionManager, prove_read,
 	ChangesTrieRootsStorage, ChangesTrieStorage,
-	key_changes, key_changes_proof, OverlayedChanges
+	key_changes, key_changes_proof, OverlayedChanges,
 };
 
 use crate::backend::{self, BlockImportOperation, PrunableStateChangesTrieStorage};
@@ -72,6 +72,9 @@ pub type ImportNotifications<Block> = mpsc::UnboundedReceiver<BlockImportNotific
 /// A stream of block finality notifications.
 pub type FinalityNotifications<Block> = mpsc::UnboundedReceiver<FinalityNotification<Block>>;
 
+type StorageUpdate<B, Block> = <<<B as backend::Backend<Block, Blake2Hasher>>::BlockImportOperation as BlockImportOperation<Block, Blake2Hasher>>::State as state_machine::Backend<Blake2Hasher>>::Transaction;
+type ChangesUpdate = trie::MemoryDB<Blake2Hasher>;
+
 /// Substrate Client
 pub struct Client<B, E, Block, RA> where Block: BlockT {
 	backend: Arc<B>,
@@ -623,40 +626,10 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
 		}
 
 		let mut transaction = self.backend.begin_operation(BlockId::Hash(parent_hash))?;
-		let (storage_update, changes_update, storage_changes) = match transaction.state()? {
-			Some(transaction_state) => {
-				let mut overlay = Default::default();
-				let r = self.executor.call_at_state::<_, _, NeverNativeValue, fn() -> NeverNativeValue>(
-					transaction_state,
-					&mut overlay,
-					"Core_execute_block",
-					&<Block as BlockT>::new(import_headers.pre().clone(), body.clone().unwrap_or_default()).encode(),
-					match (origin, self.block_execution_strategy) {
-						(BlockOrigin::NetworkInitialSync, _) | (_, ExecutionStrategy::NativeWhenPossible) =>
-							ExecutionManager::NativeWhenPossible,
-						(_, ExecutionStrategy::AlwaysWasm) => ExecutionManager::AlwaysWasm,
-						_ => ExecutionManager::Both(|wasm_result, native_result| {
-							let header = import_headers.post();
-							warn!("Consensus error between wasm and native block execution at block {}", hash);
-							warn!("   Header {:?}", header);
-							warn!("   Native result {:?}", native_result);
-							warn!("   Wasm result {:?}", wasm_result);
-							telemetry!("block.execute.consensus_failure";
-								"hash" => ?hash,
-								"origin" => ?origin,
-								"header" => ?header
-							);
-							wasm_result
-						}),
-					},
-					None,
-				);
-				let (_, storage_update, changes_update) = r?;
-				overlay.commit_prospective();
-				(Some(storage_update), Some(changes_update), Some(overlay.into_committed().collect()))
-			},
-			None => (None, None, None)
-		};
+
+		// TODO: correct path logic for when to execute this function
+		// https://github.com/paritytech/substrate/issues/1232
+		let (storage_update,changes_update,storage_changes) = self.block_execution(&import_headers, origin, hash, body.clone(), &transaction)?;
 
 		// TODO: non longest-chain rule.
 		let is_new_best = finalized || match fork_choice {
@@ -727,6 +700,58 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
 		Ok(ImportResult::Queued)
 	}
 
+	fn block_execution(
+		&self,
+		import_headers: &PrePostHeader<Block::Header>,
+		origin: BlockOrigin,
+		hash: Block::Hash,
+		body: Option<Vec<Block::Extrinsic>>,
+		transaction: &B::BlockImportOperation,
+	) -> error::Result<(
+		Option<StorageUpdate<B, Block>>, 
+		Option<Option<ChangesUpdate>>, 
+		Option<Vec<(Vec<u8>, Option<Vec<u8>>)>>,
+	)> 
+		where
+			E: CallExecutor<Block, Blake2Hasher> + Send + Sync + Clone,
+	{
+		match transaction.state()? {
+			Some(transaction_state) => {
+				let mut overlay = Default::default();
+				let (_, storage_update, changes_update) = self.executor.call_at_state::<_, _, NeverNativeValue, fn() -> NeverNativeValue>(
+					transaction_state,
+					&mut overlay,
+					"Core_execute_block",
+					&<Block as BlockT>::new(import_headers.pre().clone(), body.unwrap_or_default()).encode(),
+					match (origin, self.block_execution_strategy) {
+						(BlockOrigin::NetworkInitialSync, _) | (_, ExecutionStrategy::NativeWhenPossible) =>
+							ExecutionManager::NativeWhenPossible,
+						(_, ExecutionStrategy::AlwaysWasm) => ExecutionManager::AlwaysWasm,
+						_ => ExecutionManager::Both(|wasm_result, native_result| {
+							let header = import_headers.post();
+							warn!("Consensus error between wasm and native block execution at block {}", hash);
+							warn!("   Header {:?}", header);
+							warn!("   Native result {:?}", native_result);
+							warn!("   Wasm result {:?}", wasm_result);
+							telemetry!("block.execute.consensus_failure";
+								"hash" => ?hash,
+								"origin" => ?origin,
+								"header" => ?header
+							);
+							wasm_result
+						}),
+					},
+					None,
+				)?;
+
+				overlay.commit_prospective();
+
+				Ok((Some(storage_update), Some(changes_update), Some(overlay.into_committed().collect())))
+			},
+			None => Ok((None, None, None))
+		}
+	}
+
 	/// Finalizes all blocks up to given. If a justification is provided it is
 	/// stored with the given finalized block (any other finalized blocks are
 	/// left unjustified).