chainHead: Track reported blocks to capture notification gaps

prdoc/pr_5856.prdoc

@@ -0,0 +1,17 @@
+# Schema: Polkadot SDK PRDoc Schema (prdoc) v1.0.0
+# See doc at https://raw.githubusercontent.com/paritytech/polkadot-sdk/master/prdoc/schema_user.json
+
+title: Extend state tracking of chainHead to capture notification gaps
+
+doc:
+  - audience: Node Dev
+    description: |
+      This PR extends the state tracking of the RPC-v2 chainHead methods.
+      ChainHead tracks the reported blocks to detect notification gaps.
+      This state tracking ensures we can detect `NewBlock` events for
+      which we did not report previously the parent hash.
+
+crates:
+  - name: sc-rpc-spec-v2
+    bump: minor
+

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

@@ -24,7 +24,7 @@ use crate::chain_head::{
 		BestBlockChanged, Finalized, FollowEvent, Initialized, NewBlock, RuntimeEvent,
 		RuntimeVersionEvent,
 	},
-	subscription::{SubscriptionManagement, SubscriptionManagementError},
+	subscription::{InsertedSubscriptionData, SubscriptionManagement, SubscriptionManagementError},
 };
 use futures::{
 	channel::oneshot,
@@ -53,8 +53,6 @@ use std::{
 /// `Initialized` event.
 const MAX_FINALIZED_BLOCKS: usize = 16;
 
-use super::subscription::InsertedSubscriptionData;
-
 /// Generates the events of the `chainHead_follow` method.
 pub struct ChainHeadFollower<BE: Backend<Block>, Block: BlockT, Client> {
 	/// Substrate client.
@@ -71,6 +69,8 @@ pub struct ChainHeadFollower<BE: Backend<Block>, Block: BlockT, Client> {
 	current_best_block: Option<Block::Hash>,
 	/// LRU cache of pruned blocks.
 	pruned_blocks: LruMap<Block::Hash, ()>,
+	/// LRU cache of pruned blocks.
+	announced_blocks: LruMap<Block::Hash, ()>,
 	/// Stop all subscriptions if the distance between the leaves and the current finalized
 	/// block is larger than this value.
 	max_lagging_distance: usize,
@@ -96,6 +96,9 @@ impl<BE: Backend<Block>, Block: BlockT, Client> ChainHeadFollower<BE, Block, Cli
 			pruned_blocks: LruMap::new(ByLength::new(
 				MAX_PINNED_BLOCKS.try_into().unwrap_or(u32::MAX),
 			)),
+			announced_blocks: LruMap::new(ByLength::new(
+				MAX_PINNED_BLOCKS.try_into().unwrap_or(u32::MAX),
+			)),
 			max_lagging_distance,
 		}
 	}
@@ -214,7 +217,7 @@ where
 	///
 	/// 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(
+	fn distance_within_reason(
 		&self,
 		block: Block::Hash,
 		finalized: Block::Hash,
@@ -250,7 +253,7 @@ where
 		// 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)?;
+			self.distance_within_reason(*leaf, finalized)?;
 		}
 
 		for leaf in leaves {
@@ -326,6 +329,10 @@ where
 		let finalized_block_hash = startup_point.finalized_hash;
 		let finalized_block_runtime = self.generate_runtime_event(finalized_block_hash, None);
 
+		for finalized in &finalized_block_hashes {
+			self.announced_blocks.insert(*finalized, ());
+		}
+
 		let initialized_event = FollowEvent::Initialized(Initialized {
 			finalized_block_hashes: finalized_block_hashes.into(),
 			finalized_block_runtime,
@@ -336,6 +343,13 @@ where
 
 		finalized_block_descendants.push(initialized_event);
 		for (child, parent) in initial_blocks.into_iter() {
+			// If the parent was not announced we have a gap currently.
+			// This can happen during a WarpSync.
+			if self.announced_blocks.get(&parent).is_none() {
+				return Err(SubscriptionManagementError::BlockHeaderAbsent);
+			}
+			self.announced_blocks.insert(child, ());
+
 			let new_runtime = self.generate_runtime_event(child, Some(parent));
 
 			let event = FollowEvent::NewBlock(NewBlock {
@@ -351,6 +365,11 @@ where
 		// Generate a new best block event.
 		let best_block_hash = startup_point.best_hash;
 		if best_block_hash != finalized_block_hash {
+			if self.announced_blocks.get(&best_block_hash).is_none() {
+				return Err(SubscriptionManagementError::BlockHeaderAbsent);
+			}
+			self.announced_blocks.insert(best_block_hash, ());
+
 			let best_block = FollowEvent::BestBlockChanged(BestBlockChanged { best_block_hash });
 			self.current_best_block = Some(best_block_hash);
 			finalized_block_descendants.push(best_block);
@@ -408,21 +427,30 @@ where
 		notification: BlockImportNotification<Block>,
 		startup_point: &StartupPoint<Block>,
 	) -> Result<Vec<FollowEvent<Block::Hash>>, SubscriptionManagementError> {
-		// The block was already pinned by the initial block events or by the finalized event.
-		if !self.sub_handle.pin_block(&self.sub_id, notification.hash)? {
-			return Ok(Default::default())
-		}
+		let block_hash = notification.hash;
+
+		// Ensure the block is pinned before generating the events.
+		self.sub_handle.pin_block(&self.sub_id, block_hash)?;
 
 		// Ensure we are only reporting blocks after the starting point.
 		if *notification.header.number() < startup_point.finalized_number {
 			return Ok(Default::default())
 		}
 
-		Ok(self.generate_import_events(
-			notification.hash,
-			*notification.header.parent_hash(),
-			notification.is_new_best,
-		))
+		if self.announced_blocks.get(&block_hash).is_some() {
+			// Block was already reported by the finalized branch.
+			return Ok(Default::default())
+		}
+
+		// Double check the parent hash. If the parent hash is not reported, we have a gap.
+		let parent_block_hash = *notification.header.parent_hash();
+		if self.announced_blocks.get(&parent_block_hash).is_none() {
+			// The parent block was not reported, we have a gap.
+			return Err(SubscriptionManagementError::Custom("Parent block was not reported".into()))
+		}
+
+		self.announced_blocks.insert(block_hash, ());
+		Ok(self.generate_import_events(block_hash, parent_block_hash, notification.is_new_best))
 	}
 
 	/// Generates new block events from the given finalized hashes.
@@ -448,19 +476,29 @@ where
 			return Err(SubscriptionManagementError::BlockHeaderAbsent)
 		};
 
+		if self.announced_blocks.get(first_header.parent_hash()).is_none() {
+			return Err(SubscriptionManagementError::Custom(
+				"Parent block was not reported for a finalized block".into(),
+			));
+		}
+
 		let parents =
 			std::iter::once(first_header.parent_hash()).chain(finalized_block_hashes.iter());
 		for (i, (hash, parent)) in finalized_block_hashes.iter().zip(parents).enumerate() {
-			// Check if the block was already reported and thus, is already pinned.
-			if !self.sub_handle.pin_block(&self.sub_id, *hash)? {
-				continue
+			// Ensure the block is pinned before generating the events.
+			self.sub_handle.pin_block(&self.sub_id, *hash)?;
+
+			// Check if the block was already reported.
+			if self.announced_blocks.get(hash).is_some() {
+				continue;
 			}
 
 			// Generate `NewBlock` events for all blocks beside the last block in the list
 			let is_last = i + 1 == finalized_block_hashes.len();
 			if !is_last {
 				// Generate only the `NewBlock` event for this block.
 				events.extend(self.generate_import_events(*hash, *parent, false));
+				self.announced_blocks.insert(*hash, ());
 				continue;
 			}
 
@@ -483,7 +521,8 @@ where
 			}
 
 			// Let's generate the `NewBlock` and `NewBestBlock` events for the block.
-			events.extend(self.generate_import_events(*hash, *parent, true))
+			events.extend(self.generate_import_events(*hash, *parent, true));
+			self.announced_blocks.insert(*hash, ());
 		}
 
 		Ok(events)
@@ -545,6 +584,10 @@ where
 		let pruned_block_hashes =
 			self.get_pruned_hashes(&notification.stale_heads, last_finalized)?;
 
+		for finalized in &finalized_block_hashes {
+			self.announced_blocks.insert(*finalized, ());
+		}
+
 		let finalized_event = FollowEvent::Finalized(Finalized {
 			finalized_block_hashes,
 			pruned_block_hashes: pruned_block_hashes.clone(),

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

@@ -4052,3 +4052,72 @@ async fn follow_report_best_block_of_a_known_block() {
 	});
 	assert_eq!(event, expected);
 }
+
+#[tokio::test]
+async fn follow_event_with_unknown_parent() {
+	let builder = TestClientBuilder::new();
+	let backend = builder.backend();
+	let client = Arc::new(builder.build());
+
+	let client_mock = Arc::new(ChainHeadMockClient::new(client.clone()));
+
+	let api = ChainHead::new(
+		client_mock.clone(),
+		backend,
+		Arc::new(TaskExecutor::default()),
+		ChainHeadConfig {
+			global_max_pinned_blocks: MAX_PINNED_BLOCKS,
+			subscription_max_pinned_duration: Duration::from_secs(MAX_PINNED_SECS),
+			subscription_max_ongoing_operations: MAX_OPERATIONS,
+			operation_max_storage_items: MAX_PAGINATION_LIMIT,
+			max_lagging_distance: MAX_LAGGING_DISTANCE,
+			max_follow_subscriptions_per_connection: MAX_FOLLOW_SUBSCRIPTIONS_PER_CONNECTION,
+		},
+	)
+	.into_rpc();
+
+	let finalized_hash = client.info().finalized_hash;
+	let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap();
+	// Initialized must always be reported first.
+	let event: FollowEvent<String> = get_next_event(&mut sub).await;
+	let expected = FollowEvent::Initialized(Initialized {
+		finalized_block_hashes: vec![format!("{:?}", finalized_hash)],
+		finalized_block_runtime: None,
+		with_runtime: false,
+	});
+	assert_eq!(event, expected);
+
+	// Block tree:
+	//
+	// finalized -> (gap: block 1) -> block 2
+	//
+	// Block 1 is not announced yet. ChainHead should report the stop
+	// event when encountering an unknown parent of block 2.
+
+	// Note: `client` is used just for constructing the blocks.
+	// The blocks are imported to chainHead using the `client_mock`.
+	let block_1 = BlockBuilderBuilder::new(&*client)
+		.on_parent_block(client.chain_info().genesis_hash)
+		.with_parent_block_number(0)
+		.build()
+		.unwrap()
+		.build()
+		.unwrap()
+		.block;
+	let block_1_hash = block_1.hash();
+	client.import(BlockOrigin::Own, block_1.clone()).await.unwrap();
+
+	let block_2 = BlockBuilderBuilder::new(&*client)
+		.on_parent_block(block_1_hash)
+		.with_parent_block_number(1)
+		.build()
+		.unwrap()
+		.build()
+		.unwrap()
+		.block;
+	client.import(BlockOrigin::Own, block_2.clone()).await.unwrap();
+
+	run_with_timeout(client_mock.trigger_import_stream(block_2.header)).await;
+	// When importing the block 2, chainHead detects a gap in our blocks and stops.
+	assert_matches!(get_next_event::<FollowEvent<String>>(&mut sub).await, FollowEvent::Stop);
+}