fix: chain sel + av-store error handling (#4159)

* fix: error handling

* more error handling

* fixup

* fixup

* fixup name

* fixup

* simplify

* spelling, docs

polkadot/node/core/av-store/src/lib.rs

@@ -162,9 +162,9 @@ fn query_inner<D: Decode>(
 			Ok(Some(res))
 		},
 		Ok(None) => Ok(None),
-		Err(e) => {
-			tracing::warn!(target: LOG_TARGET, err = ?e, "Error reading from the availability store");
-			Err(e.into())
+		Err(err) => {
+			tracing::warn!(target: LOG_TARGET, ?err, "Error reading from the availability store");
+			Err(err.into())
 		},
 	}
 }
@@ -365,6 +365,20 @@ pub enum Error {
 	CustomDatabase,
 }
 
+impl Error {
+	/// Determine if the error is irrecoverable
+	/// or notifying the user via means of logging
+	/// is sufficient.
+	fn is_fatal(&self) -> bool {
+		match self {
+			Self::Io(_) => true,
+			Self::Oneshot(_) => true,
+			Self::CustomDatabase => true,
+			_ => false,
+		}
+	}
+}
+
 impl Error {
 	fn trace(&self) {
 		match self {
@@ -524,8 +538,7 @@ where
 		match res {
 			Err(e) => {
 				e.trace();
-
-				if let Error::Subsystem(SubsystemError::Context(_)) = e {
+				if e.is_fatal() {
 					break
 				}
 			},
@@ -840,8 +853,18 @@ where
 			let (tx, rx) = oneshot::channel();
 			ctx.send_message(ChainApiMessage::FinalizedBlockHash(batch_num, tx)).await;
 
-			match rx.await?? {
-				None => {
+			match rx.await? {
+				Err(err) => {
+					tracing::warn!(
+						target: LOG_TARGET,
+						batch_num,
+						?err,
+						"Failed to retrieve finalized block number.",
+					);
+
+					break
+				},
+				Ok(None) => {
 					tracing::warn!(
 						target: LOG_TARGET,
 						"Availability store was informed that block #{} is finalized, \
@@ -851,7 +874,7 @@ where
 
 					break
 				},
-				Some(h) => h,
+				Ok(Some(h)) => h,
 			}
 		};
 
@@ -1093,7 +1116,7 @@ fn process_message(
 				},
 				Err(e) => {
 					let _ = tx.send(Err(()));
-					return Err(e)
+					return Err(e.into())
 				},
 			}
 		},

polkadot/node/core/chain-selection/src/lib.rs

@@ -348,14 +348,11 @@ async fn run<Context, B>(
 	B: Backend,
 {
 	loop {
-		let res = run_iteration(&mut ctx, &mut backend, &stagnant_check_interval, &*clock).await;
+		let res = run_until_error(&mut ctx, &mut backend, &stagnant_check_interval, &*clock).await;
 		match res {
 			Err(e) => {
 				e.trace();
-
-				if let Error::Subsystem(SubsystemError::Context(_)) = e {
-					break
-				}
+				break
 			},
 			Ok(()) => {
 				tracing::info!(target: LOG_TARGET, "received `Conclude` signal, exiting");
@@ -370,7 +367,7 @@ async fn run<Context, B>(
 //
 // A return value of `Ok` indicates that an exit should be made, while non-fatal errors
 // lead to another call to this function.
-async fn run_iteration<Context, B>(
+async fn run_until_error<Context, B>(
 	ctx: &mut Context,
 	backend: &mut B,
 	stagnant_check_interval: &StagnantCheckInterval,
@@ -440,21 +437,36 @@ async fn fetch_finalized(
 	ctx: &mut impl SubsystemContext,
 ) -> Result<Option<(Hash, BlockNumber)>, Error> {
 	let (number_tx, number_rx) = oneshot::channel();
-	let (hash_tx, hash_rx) = oneshot::channel();
 
 	ctx.send_message(ChainApiMessage::FinalizedBlockNumber(number_tx)).await;
 
-	let number = number_rx.await??;
+	let number = match number_rx.await? {
+		Ok(number) => number,
+		Err(err) => {
+			tracing::warn!(target: LOG_TARGET, ?err, "Fetching finalized number failed");
+			return Ok(None)
+		},
+	};
+
+	let (hash_tx, hash_rx) = oneshot::channel();
 
 	ctx.send_message(ChainApiMessage::FinalizedBlockHash(number, hash_tx)).await;
 
-	match hash_rx.await?? {
-		None => {
-			tracing::warn!(target: LOG_TARGET, number, "Missing hash for finalized block number");
-
-			return Ok(None)
+	match hash_rx.await? {
+		Err(err) => {
+			tracing::warn!(
+				target: LOG_TARGET,
+				number,
+				?err,
+				"Fetching finalized block number failed"
+			);
+			Ok(None)
 		},
-		Some(h) => Ok(Some((h, number))),
+		Ok(None) => {
+			tracing::warn!(target: LOG_TARGET, number, "Missing hash for finalized block number");
+			Ok(None)
+		},
+		Ok(Some(h)) => Ok(Some((h, number))),
 	}
 }
 
@@ -462,10 +474,13 @@ async fn fetch_header(
 	ctx: &mut impl SubsystemContext,
 	hash: Hash,
 ) -> Result<Option<Header>, Error> {
-	let (h_tx, h_rx) = oneshot::channel();
-	ctx.send_message(ChainApiMessage::BlockHeader(hash, h_tx)).await;
+	let (tx, rx) = oneshot::channel();
+	ctx.send_message(ChainApiMessage::BlockHeader(hash, tx)).await;
 
-	h_rx.await?.map_err(Into::into)
+	Ok(rx.await?.unwrap_or_else(|err| {
+		tracing::warn!(target: LOG_TARGET, ?hash, ?err, "Missing hash for finalized block number");
+		None
+	}))
 }
 
 async fn fetch_block_weight(
@@ -475,7 +490,12 @@ async fn fetch_block_weight(
 	let (tx, rx) = oneshot::channel();
 	ctx.send_message(ChainApiMessage::BlockWeight(hash, tx)).await;
 
-	rx.await?.map_err(Into::into)
+	let res = rx.await?;
+
+	Ok(res.unwrap_or_else(|err| {
+		tracing::warn!(target: LOG_TARGET, ?hash, ?err, "Missing hash for finalized block number");
+		None
+	}))
 }
 
 // Handle a new active leaf.
@@ -590,7 +610,7 @@ fn extract_reversion_logs(header: &Header) -> Vec<BlockNumber> {
 	logs
 }
 
-// Handle a finalized block event.
+/// Handle a finalized block event.
 fn handle_finalized_block(
 	backend: &mut impl Backend,
 	finalized_hash: Hash,