Improve error handling and move logging outside of interface

client/consensus/common/src/parachain_consensus.rs

@@ -15,7 +15,7 @@
 // along with Cumulus.  If not, see <http://www.gnu.org/licenses/>.
 
 use async_trait::async_trait;
-use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface};
+use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult};
 use sc_client_api::{
 	Backend, BlockBackend, BlockImportNotification, BlockchainEvents, Finalizer, UsageProvider,
 };
@@ -34,6 +34,8 @@ use futures::{select, FutureExt, Stream, StreamExt};
 
 use std::{pin::Pin, sync::Arc};
 
+const LOG_TARGET: &str = "cumulus-consensus";
+
 /// Helper for the relay chain client. This is expected to be a lightweight handle like an `Arc`.
 #[async_trait]
 pub trait RelaychainClient: Clone + 'static {
@@ -44,10 +46,10 @@ pub trait RelaychainClient: Clone + 'static {
 	type HeadStream: Stream<Item = Vec<u8>> + Send + Unpin;
 
 	/// Get a stream of new best heads for the given parachain.
-	async fn new_best_heads(&self, para_id: ParaId) -> Self::HeadStream;
+	async fn new_best_heads(&self, para_id: ParaId) -> RelayChainResult<Self::HeadStream>;
 
 	/// Get a stream of finalized heads for the given parachain.
-	async fn finalized_heads(&self, para_id: ParaId) -> Self::HeadStream;
+	async fn finalized_heads(&self, para_id: ParaId) -> RelayChainResult<Self::HeadStream>;
 
 	/// Returns the parachain head for the given `para_id` at the given block id.
 	async fn parachain_head_at(
@@ -68,7 +70,13 @@ where
 	R: RelaychainClient,
 	B: Backend<Block>,
 {
-	let mut finalized_heads = relay_chain.finalized_heads(para_id).await;
+	let mut finalized_heads = match relay_chain.finalized_heads(para_id).await {
+		Ok(finalized_heads_stream) => finalized_heads_stream,
+		Err(err) => {
+			tracing::error!(target: LOG_TARGET, error = ?err, "Unable to retrieve finalized heads stream.");
+			return
+		},
+	};
 
 	loop {
 		let finalized_head = if let Some(h) = finalized_heads.next().await {
@@ -167,7 +175,14 @@ async fn follow_new_best<P, R, Block, B>(
 	R: RelaychainClient,
 	B: Backend<Block>,
 {
-	let mut new_best_heads = relay_chain.new_best_heads(para_id).await.fuse();
+	let mut new_best_heads = match relay_chain.new_best_heads(para_id).await {
+		Ok(best_heads_stream) => best_heads_stream.fuse(),
+		Err(err) => {
+			tracing::error!(target: LOG_TARGET, error = ?err, "Unable to retrieve best heads stream.");
+			return
+		},
+	};
+
 	let mut imported_blocks = parachain.import_notification_stream().fuse();
 	// The unset best header of the parachain. Will be `Some(_)` when we have imported a relay chain
 	// block before the parachain block it included. In this case we need to wait for this block to
@@ -379,13 +394,12 @@ where
 
 	type HeadStream = Pin<Box<dyn Stream<Item = Vec<u8>> + Send>>;
 
-	async fn new_best_heads(&self, para_id: ParaId) -> Self::HeadStream {
+	async fn new_best_heads(&self, para_id: ParaId) -> RelayChainResult<Self::HeadStream> {
 		let relay_chain = self.clone();
 
-		//TODO: Error handling
-		self.new_best_notification_stream()
-			.await
-			.expect("")
+		let new_best_notification_stream = self
+			.new_best_notification_stream()
+			.await?
 			.filter_map(move |n| {
 				let relay_chain = relay_chain.clone();
 				async move {
@@ -396,15 +410,16 @@ where
 						.flatten()
 				}
 			})
-			.boxed()
+			.boxed();
+		Ok(new_best_notification_stream)
 	}
 
-	async fn finalized_heads(&self, para_id: ParaId) -> Self::HeadStream {
+	async fn finalized_heads(&self, para_id: ParaId) -> RelayChainResult<Self::HeadStream> {
 		let relay_chain = self.clone();
 
-		self.finality_notification_stream()
-			.await
-			.expect("")
+		let finality_notification_stream = self
+			.finality_notification_stream()
+			.await?
 			.filter_map(move |n| {
 				let relay_chain = relay_chain.clone();
 				async move {
@@ -415,7 +430,8 @@ where
 						.flatten()
 				}
 			})
-			.boxed()
+			.boxed();
+		Ok(finality_notification_stream)
 	}
 
 	async fn parachain_head_at(

client/consensus/common/src/tests.rs

@@ -18,7 +18,7 @@ use crate::*;
 
 use async_trait::async_trait;
 use codec::Encode;
-use cumulus_relay_chain_interface::RelayChainError;
+use cumulus_relay_chain_interface::{RelayChainError, RelayChainResult};
 use cumulus_test_client::{
 	runtime::{Block, Header},
 	Backend, Client, InitBlockBuilder, TestClientBuilder, TestClientBuilderExt,
@@ -74,7 +74,7 @@ impl crate::parachain_consensus::RelaychainClient for Relaychain {
 
 	type HeadStream = Box<dyn Stream<Item = Vec<u8>> + Send + Unpin>;
 
-	async fn new_best_heads(&self, _: ParaId) -> Self::HeadStream {
+	async fn new_best_heads(&self, _: ParaId) -> RelayChainResult<Self::HeadStream> {
 		let stream = self
 			.inner
 			.lock()
@@ -83,10 +83,10 @@ impl crate::parachain_consensus::RelaychainClient for Relaychain {
 			.take()
 			.expect("Should only be called once");
 
-		Box::new(stream.map(|v| v.encode()))
+		Ok(Box::new(stream.map(|v| v.encode())))
 	}
 
-	async fn finalized_heads(&self, _: ParaId) -> Self::HeadStream {
+	async fn finalized_heads(&self, _: ParaId) -> RelayChainResult<Self::HeadStream> {
 		let stream = self
 			.inner
 			.lock()
@@ -95,7 +95,7 @@ impl crate::parachain_consensus::RelaychainClient for Relaychain {
 			.take()
 			.expect("Should only be called once");
 
-		Box::new(stream.map(|v| v.encode()))
+		Ok(Box::new(stream.map(|v| v.encode())))
 	}
 
 	async fn parachain_head_at(

client/consensus/relay-chain/src/lib.rs

@@ -176,7 +176,7 @@ where
 			.propose(
 				inherent_data,
 				Default::default(),
-				//TODO: Fix this.
+				// TODO: Fix this.
 				Duration::from_millis(500),
 				// Set the block limit to 50% of the maximum PoV size.
 				//

client/network/src/tests.rs

@@ -91,7 +91,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 		self.relay_backend
 			.blockchain()
 			.status(block_id)
-			.map_err(|err| RelayChainError::BlockchainError(err.to_string()))
+			.map_err(RelayChainError::BlockchainError)
 	}
 
 	async fn best_block_hash(&self) -> RelayChainResult<PHash> {
@@ -206,7 +206,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 		&self,
 		_: &polkadot_service::BlockId,
 		_: &Vec<Vec<u8>>,
-	) -> Result<Option<sc_client_api::StorageProof>, RelayChainError> {
+	) -> Result<sc_client_api::StorageProof, RelayChainError> {
 		unimplemented!("Not needed for test")
 	}
 

client/relay-chain-interface/src/lib.rs

@@ -52,8 +52,10 @@ pub enum RelayChainError {
 	)]
 	WaitBlockchainError(PHash, sp_blockchain::Error),
 	#[display(fmt = "Blockchain returned an error: {:?}", _0)]
-	BlockchainError(String),
+	BlockchainError(sp_blockchain::Error),
 	StateMachineError(String),
+	#[display(fmt = "Unspecified error occured: {:?}", _0)]
+	GenericError(String),
 }
 impl std::error::Error for RelayChainError {}
 
@@ -160,7 +162,7 @@ pub trait RelayChainInterface: Send + Sync {
 		&self,
 		block_id: &BlockId,
 		relevant_keys: &Vec<Vec<u8>>,
-	) -> Result<Option<StorageProof>, RelayChainError>;
+	) -> Result<StorageProof, RelayChainError>;
 }
 
 #[async_trait]
@@ -254,7 +256,7 @@ where
 		&self,
 		block_id: &BlockId,
 		relevant_keys: &Vec<Vec<u8>>,
-	) -> RelayChainResult<Option<StorageProof>> {
+	) -> RelayChainResult<StorageProof> {
 		(**self).prove_read(block_id, relevant_keys).await
 	}
 

client/relay-chain-local/src/lib.rs

@@ -42,7 +42,6 @@ use sp_consensus::SyncOracle;
 use sp_core::{sp_std::collections::btree_map::BTreeMap, Pair};
 use sp_state_machine::{Backend as StateBackend, StorageValue};
 
-const LOG_TARGET: &str = "relay-chain-local";
 /// The timeout in seconds after that the waiting for a block should be aborted.
 const TIMEOUT_IN_SECONDS: u64 = 6;
 
@@ -100,15 +99,7 @@ where
 				sp_core::ExecutionContext::Importing,
 				para_id,
 			)
-			.map_err(|e| {
-				tracing::error!(
-					target: LOG_TARGET,
-					relay_parent = ?relay_parent,
-					error = ?e,
-					"An error occured during requesting the downward messages.",
-				);
-				RelayChainError::ApiError(e)
-			})
+			.map_err(RelayChainError::ApiError)
 	}
 
 	async fn retrieve_all_inbound_hrmp_channel_contents(
@@ -123,15 +114,7 @@ where
 				sp_core::ExecutionContext::Importing,
 				para_id,
 			)
-			.map_err(|e| {
-				tracing::error!(
-					target: LOG_TARGET,
-					relay_parent = ?relay_parent,
-					error = ?e,
-					"An error occured during requesting the inbound HRMP messages.",
-				);
-				RelayChainError::ApiError(e)
-			})
+			.map_err(RelayChainError::ApiError)
 	}
 
 	async fn persisted_validation_data(
@@ -202,7 +185,7 @@ where
 		self.backend
 			.blockchain()
 			.status(block_id)
-			.map_err(|err| RelayChainError::BlockchainError(err.to_string()))
+			.map_err(RelayChainError::BlockchainError)
 	}
 
 	async fn is_major_syncing(&self) -> RelayChainResult<bool> {
@@ -219,45 +202,20 @@ where
 		block_id: &BlockId,
 		key: &[u8],
 	) -> Result<Option<StorageValue>, RelayChainError> {
-		let state = self
-			.backend
-			.state_at(*block_id)
-			.map_err(|err| RelayChainError::BlockchainError(err.to_string()))?;
-		state.storage(key).map_err(RelayChainError::BlockchainError)
+		let state = self.backend.state_at(*block_id).map_err(RelayChainError::BlockchainError)?;
+		state.storage(key).map_err(RelayChainError::GenericError)
 	}
 
 	async fn prove_read(
 		&self,
 		block_id: &BlockId,
 		relevant_keys: &Vec<Vec<u8>>,
-	) -> Result<Option<StorageProof>, RelayChainError> {
-		let state_backend = self
-			.backend
-			.state_at(*block_id)
-			.map_err(|e| {
-				tracing::error!(
-					target: LOG_TARGET,
-					relay_parent = ?block_id,
-					error = ?e,
-					"Cannot obtain the state of the relay chain.",
-				);
-			})
-			.ok();
-
-		match state_backend {
-			Some(state) => sp_state_machine::prove_read(state, relevant_keys)
-				.map_err(|e| {
-					tracing::error!(
-						target: LOG_TARGET,
-						relay_parent = ?block_id,
-						error = ?e,
-						"Failed to collect required relay chain state storage proof.",
-					);
-					RelayChainError::StateMachineError(e.to_string())
-				})
-				.map(Some),
-			None => Ok(None),
-		}
+	) -> Result<StorageProof, RelayChainError> {
+		let state_backend =
+			self.backend.state_at(*block_id).map_err(RelayChainError::BlockchainError)?;
+
+		sp_state_machine::prove_read(state_backend, relevant_keys)
+			.map_err(|e| RelayChainError::StateMachineError(e.to_string()))
 	}
 
 	/// Wait for a given relay chain block in an async way.
@@ -337,7 +295,7 @@ where
 	let block_id = BlockId::Hash(hash);
 	match backend.blockchain().status(block_id) {
 		Ok(BlockStatus::InChain) => return Ok(BlockCheckStatus::InChain),
-		Err(err) => return Err(RelayChainError::BlockchainError(err.to_string())),
+		Err(err) => return Err(RelayChainError::BlockchainError(err)),
 		_ => {},
 	}
 

client/service/src/lib.rs

@@ -107,7 +107,6 @@ where
 		.spawn_essential_handle()
 		.spawn("cumulus-consensus", None, consensus);
 
-	//TODO: error handling
 	let pov_recovery = cumulus_client_pov_recovery::PoVRecovery::new(
 		relay_chain_interface
 			.overseer_handle()
@@ -125,7 +124,6 @@ where
 		.spawn_essential_handle()
 		.spawn("cumulus-pov-recovery", None, pov_recovery.run());
 
-	//TODO: error handling
 	cumulus_client_collator::start_collator(cumulus_client_collator::StartCollatorParams {
 		runtime_api: client.clone(),
 		block_status,

pallets/parachain-system/src/lib.rs

@@ -603,7 +603,7 @@ pub mod pallet {
 	#[pallet::genesis_build]
 	impl<T: Config> GenesisBuild<T> for GenesisConfig {
 		fn build(&self) {
-			//TODO: Remove after https://github.com/paritytech/cumulus/issues/479
+			// TODO: Remove after https://github.com/paritytech/cumulus/issues/479
 			sp_io::storage::set(b":c", &[]);
 		}
 	}

primitives/parachain-inherent/src/client_side.rs

@@ -113,7 +113,15 @@ async fn collect_relay_storage_proof(
 	relay_chain_interface
 		.prove_read(&relay_parent_block_id, &relevant_keys)
 		.await
-		.ok()?
+		.map_err(|e| {
+			tracing::error!(
+				target: LOG_TARGET,
+				relay_parent = ?relay_parent_block_id,
+				error = ?e,
+				"Cannot obtain read proof from relay chain.",
+			);
+		})
+		.ok()
 }
 
 impl ParachainInherentData {
@@ -129,12 +137,29 @@ impl ParachainInherentData {
 		let relay_chain_state =
 			collect_relay_storage_proof(&relay_chain_interface, para_id, relay_parent).await?;
 
-		//TODO: error handling
-		let downward_messages =
-			relay_chain_interface.retrieve_dmq_contents(para_id, relay_parent).await.ok()?;
+		let downward_messages = relay_chain_interface
+			.retrieve_dmq_contents(para_id, relay_parent)
+			.await
+			.map_err(|e| {
+				tracing::error!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent,
+					error = ?e,
+					"An error occured during requesting the downward messages.",
+				);
+			})
+			.ok()?;
 		let horizontal_messages = relay_chain_interface
 			.retrieve_all_inbound_hrmp_channel_contents(para_id, relay_parent)
 			.await
+			.map_err(|e| {
+				tracing::error!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent,
+					error = ?e,
+					"An error occured during requesting the inbound HRMP messages.",
+				);
+			})
 			.ok()?;
 
 		Some(ParachainInherentData {