chainHead: Ensure reasonable distance between leaf and finalized block

substrate/client/rpc-spec-v2/src/chain_head/chain_head.rs

@@ -62,6 +62,16 @@ pub struct ChainHeadConfig {
 	pub subscription_max_pinned_duration: Duration,
 	/// The maximum number of ongoing operations per subscription.
 	pub subscription_max_ongoing_operations: usize,
+	/// Suspend the subscriptions if the distance between the leaves and the current finalized
+	/// block is larger than this value.
+	///
+	/// Subscriptions are suspended for the `suspended_duration`.
+	pub suspend_on_lagging_distance: usize,
+	/// The amount of time for which the subscriptions are suspended.
+	///
+	/// Subscriptions are suspended when the distance between any leaf
+	/// and the finalized block is too large.
+	pub suspended_duration: Duration,
 	/// The maximum number of items reported by the `chainHead_storage` before
 	/// pagination is required.
 	pub operation_max_storage_items: usize,
@@ -86,12 +96,26 @@ const MAX_ONGOING_OPERATIONS: usize = 16;
 /// before paginations is required.
 const MAX_STORAGE_ITER_ITEMS: usize = 5;
 
+/// Suspend the subscriptions if the distance between the leaves and the current finalized
+/// block is larger than this value.
+///
+/// Subscriptions are suspended for the `suspended_duration`.
+const SUSPEND_ON_LAGGING_DISTANCE: usize = 128;
+
+/// The amount of time for which the subscriptions are suspended.
+///
+/// Subscriptions are suspended when the distance between any leaf
+/// and the finalized block is too large.
+const SUSPENDED_DURATION: Duration = Duration::from_secs(30);
+
 impl Default for ChainHeadConfig {
 	fn default() -> Self {
 		ChainHeadConfig {
 			global_max_pinned_blocks: MAX_PINNED_BLOCKS,
 			subscription_max_pinned_duration: MAX_PINNED_DURATION,
 			subscription_max_ongoing_operations: MAX_ONGOING_OPERATIONS,
+			suspended_duration: SUSPENDED_DURATION,
+			suspend_on_lagging_distance: SUSPEND_ON_LAGGING_DISTANCE,
 			operation_max_storage_items: MAX_STORAGE_ITER_ITEMS,
 		}
 	}
@@ -110,6 +134,9 @@ pub struct ChainHead<BE: Backend<Block>, Block: BlockT, Client> {
 	/// The maximum number of items reported by the `chainHead_storage` before
 	/// pagination is required.
 	operation_max_storage_items: usize,
+	/// Suspend the subscriptions if the distance between the leaves and the current finalized
+	/// block is larger than this value.
+	suspend_on_lagging_distance: usize,
 	/// Phantom member to pin the block type.
 	_phantom: PhantomData<Block>,
 }
@@ -130,9 +157,11 @@ impl<BE: Backend<Block>, Block: BlockT, Client> ChainHead<BE, Block, Client> {
 				config.global_max_pinned_blocks,
 				config.subscription_max_pinned_duration,
 				config.subscription_max_ongoing_operations,
+				config.suspended_duration,
 				backend,
 			)),
 			operation_max_storage_items: config.operation_max_storage_items,
+			suspend_on_lagging_distance: config.suspend_on_lagging_distance,
 			_phantom: PhantomData,
 		}
 	}
@@ -180,6 +209,7 @@ where
 		let subscriptions = self.subscriptions.clone();
 		let backend = self.backend.clone();
 		let client = self.client.clone();
+		let suspend_on_lagging_distance = self.suspend_on_lagging_distance;
 
 		let fut = async move {
 			let Ok(sink) = pending.accept().await else { return };
@@ -189,9 +219,9 @@ where
 			// Keep track of the subscription.
 			let Some(sub_data) = subscriptions.insert_subscription(sub_id.clone(), with_runtime)
 			else {
-				// Inserting the subscription can only fail if the JsonRPSee
-				// generated a duplicate subscription ID.
-				debug!(target: LOG_TARGET, "[follow][id={:?}] Subscription already accepted", sub_id);
+				// Inserting the subscription can only fail if the JsonRPSee generated a duplicate
+				// subscription ID; or subscriptions are suspended.
+				debug!(target: LOG_TARGET, "[follow][id={:?}] Subscription already accepted or suspended", sub_id);
 				let msg = to_sub_message(&sink, &FollowEvent::<String>::Stop);
 				let _ = sink.send(msg).await;
 				return
@@ -204,9 +234,14 @@ where
 				subscriptions.clone(),
 				with_runtime,
 				sub_id.clone(),
+				suspend_on_lagging_distance,
 			);
+			let result = chain_head_follow.generate_events(sink, sub_data).await;
 
-			chain_head_follow.generate_events(sink, sub_data).await;
+			if let Err(SubscriptionManagementError::BlockDistanceTooLarge) = result {
+				debug!(target: LOG_TARGET, "[follow][id={:?}] All subscriptions are suspended", sub_id);
+				subscriptions.suspend_subscriptions();
+			}
 
 			subscriptions.remove_subscription(&sub_id);
 			debug!(target: LOG_TARGET, "[follow][id={:?}] Subscription removed", sub_id);

substrate/client/rpc-spec-v2/src/chain_head/chain_head_follow.rs

@@ -41,12 +41,14 @@ use sp_api::CallApiAt;
 use sp_blockchain::{
 	Backend as BlockChainBackend, Error as BlockChainError, HeaderBackend, HeaderMetadata, Info,
 };
-use sp_runtime::traits::{Block as BlockT, Header as HeaderT, NumberFor};
+use sp_runtime::{
+	traits::{Block as BlockT, Header as HeaderT, NumberFor},
+	SaturatedConversion, Saturating,
+};
 use std::{
 	collections::{HashSet, VecDeque},
 	sync::Arc,
 };
-
 /// The maximum number of finalized blocks provided by the
 /// `Initialized` event.
 const MAX_FINALIZED_BLOCKS: usize = 16;
@@ -67,6 +69,9 @@ pub struct ChainHeadFollower<BE: Backend<Block>, Block: BlockT, Client> {
 	sub_id: String,
 	/// The best reported block by this subscription.
 	best_block_cache: Option<Block::Hash>,
+	/// Suspend the subscriptions if the distance between the leaves and the current finalized
+	/// block is larger than this value.
+	suspend_on_lagging_distance: usize,
 }
 
 impl<BE: Backend<Block>, Block: BlockT, Client> ChainHeadFollower<BE, Block, Client> {
@@ -77,8 +82,17 @@ impl<BE: Backend<Block>, Block: BlockT, Client> ChainHeadFollower<BE, Block, Cli
 		sub_handle: Arc<SubscriptionManagement<Block, BE>>,
 		with_runtime: bool,
 		sub_id: String,
+		suspend_on_lagging_distance: usize,
 	) -> Self {
-		Self { client, backend, sub_handle, with_runtime, sub_id, best_block_cache: None }
+		Self {
+			client,
+			backend,
+			sub_handle,
+			with_runtime,
+			sub_id,
+			best_block_cache: None,
+			suspend_on_lagging_distance,
+		}
 	}
 }
 
@@ -186,6 +200,35 @@ where
 		}
 	}
 
+	/// Check the distance between the provided blocks does not exceed a
+	/// a reasonable range.
+	///
+	/// When the blocks are too far apart (potentially millions of blocks):
+	///  - Tree route is expensive to calculate.
+	///  - The RPC layer will not be able to generate the `NewBlock` events for all blocks.
+	///
+	/// This edge-case can happen for parachains where the relay chain syncs slower to
+	/// the head of the chain than the parachain node that is synced already.
+	fn distace_within_reason(
+		&self,
+		block: Block::Hash,
+		finalized: Block::Hash,
+	) -> Result<(), SubscriptionManagementError> {
+		let Some(block_num) = self.client.number(block)? else {
+			return Err(SubscriptionManagementError::BlockHashAbsent)
+		};
+		let Some(finalized_num) = self.client.number(finalized)? else {
+			return Err(SubscriptionManagementError::BlockHashAbsent)
+		};
+
+		let distance: usize = block_num.saturating_sub(finalized_num).saturated_into();
+		if distance > self.suspend_on_lagging_distance {
+			return Err(SubscriptionManagementError::BlockDistanceTooLarge);
+		}
+
+		Ok(())
+	}
+
 	/// Get the in-memory blocks of the client, starting from the provided finalized hash.
 	///
 	/// The reported blocks are pinned by this function.
@@ -198,6 +241,13 @@ where
 		let mut pruned_forks = HashSet::new();
 		let mut finalized_block_descendants = Vec::new();
 		let mut unique_descendants = HashSet::new();
+
+		// Ensure all leaves are within a reasonable distance from the finalized block,
+		// before traversing the tree.
+		for leaf in &leaves {
+			self.distace_within_reason(*leaf, finalized)?;
+		}
+
 		for leaf in leaves {
 			let tree_route = sp_blockchain::tree_route(blockchain, finalized, leaf)?;
 
@@ -542,7 +592,8 @@ where
 		mut to_ignore: HashSet<Block::Hash>,
 		sink: SubscriptionSink,
 		rx_stop: oneshot::Receiver<()>,
-	) where
+	) -> Result<(), SubscriptionManagementError>
+	where
 		EventStream: Stream<Item = NotificationType<Block>> + Unpin,
 	{
 		let mut stream_item = stream.next();
@@ -571,7 +622,7 @@ where
 					);
 					let msg = to_sub_message(&sink, &FollowEvent::<String>::Stop);
 					let _ = sink.send(msg).await;
-					return
+					return Err(err)
 				},
 			};
 
@@ -586,7 +637,8 @@ where
 
 					let msg = to_sub_message(&sink, &FollowEvent::<String>::Stop);
 					let _ = sink.send(msg).await;
-					return
+					// No need to propagate this error further, the client disconnected.
+					return Ok(())
 				}
 			}
 
@@ -598,14 +650,15 @@ where
 		// or the `Stop` receiver was triggered.
 		let msg = to_sub_message(&sink, &FollowEvent::<String>::Stop);
 		let _ = sink.send(msg).await;
+		Ok(())
 	}
 
 	/// Generate the block events for the `chainHead_follow` method.
 	pub async fn generate_events(
 		&mut self,
 		sink: SubscriptionSink,
 		sub_data: InsertedSubscriptionData<Block>,
-	) {
+	) -> Result<(), SubscriptionManagementError> {
 		// Register for the new block and finalized notifications.
 		let stream_import = self
 			.client
@@ -633,7 +686,7 @@ where
 				);
 				let msg = to_sub_message(&sink, &FollowEvent::<String>::Stop);
 				let _ = sink.send(msg).await;
-				return
+				return Err(err)
 			},
 		};
 
@@ -643,6 +696,6 @@ where
 		let stream = stream::once(futures::future::ready(initial)).chain(merged);
 
 		self.submit_events(&startup_point, stream.boxed(), pruned_forks, sink, sub_data.rx_stop)
-			.await;
+			.await
 	}
 }

substrate/client/rpc-spec-v2/src/chain_head/subscription/error.rs

@@ -41,6 +41,9 @@ pub enum SubscriptionManagementError {
 	/// The unpin method was called with duplicate hashes.
 	#[error("Duplicate hashes")]
 	DuplicateHashes,
+	/// The distance between the leaves and the current finalized block is too large.
+	#[error("Distance too large")]
+	BlockDistanceTooLarge,
 	/// Custom error.
 	#[error("Subscription error {0}")]
 	Custom(String),
@@ -57,6 +60,7 @@ impl PartialEq for SubscriptionManagementError {
 			(Self::BlockHeaderAbsent, Self::BlockHeaderAbsent) |
 			(Self::SubscriptionAbsent, Self::SubscriptionAbsent) |
 			(Self::DuplicateHashes, Self::DuplicateHashes) => true,
+			(Self::BlockDistanceTooLarge, Self::BlockDistanceTooLarge) => true,
 			(Self::Custom(lhs), Self::Custom(rhs)) => lhs == rhs,
 			_ => false,
 		}