feat: ✨ extend EthereumBlockNotification with reorg info

Author: manuelmauro

client/mapping-sync/src/kv/mod.rs

@@ -36,6 +36,7 @@ use fp_consensus::{FindLogError, Hashes, Log, PostLog, PreLog};
 use fp_rpc::EthereumRuntimeRPCApi;
 
 use crate::{EthereumBlockNotification, EthereumBlockNotificationSinks, SyncStrategy};
+use worker::BestBlockInfo;
 
 pub fn sync_block<Block: BlockT, C: HeaderBackend<Block>>(
 	storage_override: Arc<dyn StorageOverride<Block>>,
@@ -155,7 +156,7 @@ pub fn sync_one_block<Block: BlockT, C, BE>(
 	pubsub_notification_sinks: Arc<
 		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
 	>,
-	best_at_import: &mut HashMap<Block::Hash, <Block::Header as HeaderT>::Number>,
+	best_at_import: &mut HashMap<Block::Hash, BestBlockInfo<Block>>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -233,10 +234,15 @@ where
 			// This avoids race conditions where the best hash may have changed
 			// between import and sync time (e.g., during rapid reorgs).
 			// Fall back to current best hash check for blocks synced during catch-up.
-			let is_new_best =
-				best_at_import.remove(&hash).is_some() || client.info().best_hash == hash;
-			sink.unbounded_send(EthereumBlockNotification { is_new_best, hash })
-				.is_ok()
+			let best_info = best_at_import.remove(&hash);
+			let is_new_best = best_info.is_some() || client.info().best_hash == hash;
+			let reorg_info = best_info.and_then(|info| info.reorg_info);
+			sink.unbounded_send(EthereumBlockNotification {
+				is_new_best,
+				hash,
+				reorg_info,
+			})
+			.is_ok()
 		} else {
 			// Remove from the pool if in major syncing.
 			false
@@ -257,7 +263,7 @@ pub fn sync_blocks<Block: BlockT, C, BE>(
 	pubsub_notification_sinks: Arc<
 		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
 	>,
-	best_at_import: &mut HashMap<Block::Hash, <Block::Header as HeaderT>::Number>,
+	best_at_import: &mut HashMap<Block::Hash, BestBlockInfo<Block>>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -286,7 +292,7 @@ where
 	// Entries for finalized blocks are no longer needed since finalized blocks
 	// cannot be reorged and their is_new_best status is irrelevant.
 	let finalized_number = client.info().finalized_number;
-	best_at_import.retain(|_, block_number| *block_number > finalized_number);
+	best_at_import.retain(|_, info| info.block_number > finalized_number);
 
 	Ok(synced_any)
 }

client/mapping-sync/src/kv/worker.rs

@@ -37,7 +37,15 @@ use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 use fc_storage::StorageOverride;
 use fp_rpc::EthereumRuntimeRPCApi;
 
-use crate::SyncStrategy;
+use crate::{ReorgInfo, SyncStrategy};
+
+/// Information tracked at import time for a block that was `is_new_best`.
+pub struct BestBlockInfo<Block: BlockT> {
+	/// The block number (for pruning purposes).
+	pub block_number: <Block::Header as HeaderT>::Number,
+	/// Reorg info if this block became best as part of a reorganization.
+	pub reorg_info: Option<ReorgInfo<Block>>,
+}
 
 pub struct MappingSyncWorker<Block: BlockT, C, BE> {
 	import_notifications: ImportNotifications<Block>,
@@ -59,11 +67,11 @@ pub struct MappingSyncWorker<Block: BlockT, C, BE> {
 		Arc<crate::EthereumBlockNotificationSinks<crate::EthereumBlockNotification<Block>>>,
 
 	/// Tracks block hashes that were `is_new_best` at the time of their import notification,
-	/// along with their block number for pruning purposes.
+	/// along with their block number for pruning purposes and optional reorg info.
 	/// This is used to correctly determine `is_new_best` when syncing blocks, avoiding race
 	/// conditions where the best hash may have changed between import and sync time.
 	/// Entries are pruned when blocks become finalized to prevent unbounded growth.
-	best_at_import: HashMap<Block::Hash, <Block::Header as HeaderT>::Number>,
+	best_at_import: HashMap<Block::Hash, BestBlockInfo<Block>>,
 }
 
 impl<Block: BlockT, C, BE> Unpin for MappingSyncWorker<Block, C, BE> {}
@@ -126,10 +134,33 @@ where
 					fire = true;
 					// Track blocks that were `is_new_best` at import time to avoid race
 					// conditions when determining `is_new_best` at sync time.
-					// We store the block number to enable pruning of old entries.
+					// We store the block number to enable pruning of old entries,
+					// and reorg info if this block became best as part of a reorg.
 					if notification.is_new_best {
-						self.best_at_import
-							.insert(notification.hash, *notification.header.number());
+						let reorg_info = notification.tree_route.as_ref().map(|tree_route| {
+							let retracted = tree_route
+								.retracted()
+								.iter()
+								.map(|hash_and_number| hash_and_number.hash)
+								.collect();
+							let enacted = tree_route
+								.enacted()
+								.iter()
+								.map(|hash_and_number| hash_and_number.hash)
+								.collect();
+							ReorgInfo {
+								common_ancestor: tree_route.common_block().hash,
+								retracted,
+								enacted,
+							}
+						});
+						self.best_at_import.insert(
+							notification.hash,
+							BestBlockInfo {
+								block_number: *notification.header.number(),
+								reorg_info,
+							},
+						);
 					}
 				}
 				Poll::Ready(None) => return Poll::Ready(None),

client/mapping-sync/src/lib.rs

@@ -34,8 +34,26 @@ pub enum SyncStrategy {
 pub type EthereumBlockNotificationSinks<T> =
 	parking_lot::Mutex<Vec<sc_utils::mpsc::TracingUnboundedSender<T>>>;
 
-#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+/// Information about a chain reorganization.
+///
+/// When a reorg occurs, this struct contains the blocks that were removed from
+/// the canonical chain (retracted) and the blocks that were added (enacted).
+/// The `common_ancestor` is the last block that remains canonical in both
+/// the old and new chains.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ReorgInfo<Block: BlockT> {
+	/// The common ancestor block hash between the old and new canonical chains.
+	pub common_ancestor: Block::Hash,
+	/// Blocks that were removed from the canonical chain (old fork).
+	pub retracted: Vec<Block::Hash>,
+	/// Blocks that were added to the canonical chain (new fork).
+	pub enacted: Vec<Block::Hash>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub struct EthereumBlockNotification<Block: BlockT> {
 	pub is_new_best: bool,
 	pub hash: Block::Hash,
+	/// Optional reorg information. Present when this block became best as part of a reorg.
+	pub reorg_info: Option<ReorgInfo<Block>>,
 }

client/mapping-sync/src/sql/mod.rs

@@ -29,17 +29,20 @@ use sp_runtime::traits::{Block as BlockT, Header as HeaderT, UniqueSaturatedInto
 // Frontier
 use fp_rpc::EthereumRuntimeRPCApi;
 
-use crate::{EthereumBlockNotification, EthereumBlockNotificationSinks, SyncStrategy};
+use crate::{EthereumBlockNotification, EthereumBlockNotificationSinks, ReorgInfo, SyncStrategy};
 
 /// Defines the commands for the sync worker.
 #[derive(Debug)]
-pub enum WorkerCommand {
+pub enum WorkerCommand<Block: BlockT> {
 	/// Resume indexing from the last indexed canon block.
 	ResumeSync,
 	/// Index leaves.
 	IndexLeaves(Vec<H256>),
 	/// Index the best block known so far via import notifications.
-	IndexBestBlock(H256),
+	IndexBestBlock {
+		block_hash: H256,
+		reorg_info: Option<ReorgInfo<Block>>,
+	},
 	/// Canonicalize the enacted and retracted blocks reported via import notifications.
 	Canonicalize {
 		common: H256,
@@ -80,7 +83,7 @@ where
 		pubsub_notification_sinks: Arc<
 			EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
 		>,
-	) -> tokio::sync::mpsc::Sender<WorkerCommand> {
+	) -> tokio::sync::mpsc::Sender<WorkerCommand<Block>> {
 		let (tx, mut rx) = tokio::sync::mpsc::channel(100);
 		tokio::task::spawn(async move {
 			while let Some(cmd) = rx.recv().await {
@@ -122,7 +125,10 @@ where
 							.await;
 						}
 					}
-					WorkerCommand::IndexBestBlock(block_hash) => {
+					WorkerCommand::IndexBestBlock {
+						block_hash,
+						reorg_info,
+					} => {
 						index_canonical_block_and_ancestors(
 							client.clone(),
 							substrate_backend.clone(),
@@ -135,6 +141,7 @@ where
 							let _ = sink.unbounded_send(EthereumBlockNotification {
 								is_new_best: true,
 								hash: block_hash,
+								reorg_info: reorg_info.clone(),
 							});
 						}
 					}
@@ -229,7 +236,8 @@ where
 						notification.is_new_best,
 					);
 					if notification.is_new_best {
-						if let Some(tree_route) = notification.tree_route {
+						// Extract reorg info from tree_route if present
+						let reorg_info = if let Some(ref tree_route) = notification.tree_route {
 							log::debug!(
 								target: "frontier-sql",
 								"🔀  Re-org happened at new best {}, proceeding to canonicalize db",
@@ -249,12 +257,23 @@ where
 							let common = tree_route.common_block().hash;
 							tx.send(WorkerCommand::Canonicalize {
 								common,
-								enacted,
-								retracted,
+								enacted: enacted.clone(),
+								retracted: retracted.clone(),
 							}).await.ok();
-						}
 
-						tx.send(WorkerCommand::IndexBestBlock(notification.hash)).await.ok();
+							Some(ReorgInfo {
+								common_ancestor: common,
+								retracted,
+								enacted,
+							})
+						} else {
+							None
+						};
+
+						tx.send(WorkerCommand::IndexBestBlock {
+							block_hash: notification.hash,
+							reorg_info,
+						}).await.ok();
 					}
 				}
 			}

client/rpc/src/eth_pubsub.rs

@@ -21,6 +21,7 @@ use std::{marker::PhantomData, sync::Arc};
 use ethereum::TransactionV3 as EthereumTransaction;
 use futures::{future, FutureExt as _, StreamExt as _};
 use jsonrpsee::{core::traits::IdProvider, server::PendingSubscriptionSink};
+use log::debug;
 // Substrate
 use sc_client_api::{
 	backend::{Backend, StorageProvider},
@@ -118,16 +119,21 @@ where
 		}
 	}
 
-	fn notify_header(
-		&self,
-		notification: EthereumBlockNotification<B>,
-	) -> future::Ready<Option<PubSubResult>> {
-		let res = if notification.is_new_best {
-			self.storage_override.current_block(notification.hash)
-		} else {
-			None
-		};
-		future::ready(res.map(PubSubResult::header))
+	/// Get headers for enacted blocks during a reorg.
+	///
+	/// Per Ethereum spec (https://github.com/ethereum/go-ethereum/wiki/RPC-PUB-SUB#newheads):
+	/// "When a chain reorganization occurs, this subscription will emit an event
+	/// containing all new headers (blocks) for the new chain. This means that you
+	/// may see multiple headers emitted with the same height (block number)."
+	///
+	/// Returns headers in ascending order (oldest first).
+	/// Note: `enacted` from tree_route already includes the new best block.
+	fn get_enacted_headers(&self, enacted: &[B::Hash]) -> Vec<PubSubResult> {
+		enacted
+			.iter()
+			.filter_map(|hash| self.storage_override.current_block(*hash))
+			.map(PubSubResult::header)
+			.collect()
 	}
 
 	fn notify_logs(
@@ -252,10 +258,45 @@ where
 		let fut = async move {
 			match kind {
 				Kind::NewHeads => {
-					let stream = block_notification_stream
-						.filter_map(move |notification| pubsub.notify_header(notification));
+					// Per Ethereum spec, when a reorg occurs, we must emit all headers
+					// for the new canonical chain. The reorg_info field in the notification
+					// contains the enacted blocks when a reorg occurred.
+					let stream = block_notification_stream.filter_map(move |notification| {
+						if !notification.is_new_best {
+							return future::ready(None);
+						}
+
+						// Check if this block came from a reorg
+						let headers = if let Some(ref reorg_info) = notification.reorg_info {
+							debug!(
+								target: "eth-pubsub",
+								"Reorg detected: {} blocks retracted, {} blocks enacted",
+								reorg_info.retracted.len(),
+								reorg_info.enacted.len()
+							);
+							// Emit all enacted blocks (already includes the new best block)
+							pubsub.get_enacted_headers(&reorg_info.enacted)
+						} else {
+							// Normal case: just emit the new block
+							if let Some(block) = pubsub.storage_override.current_block(notification.hash) {
+								vec![PubSubResult::header(block)]
+							} else {
+								return future::ready(None);
+							}
+						};
+
+						if headers.is_empty() {
+							return future::ready(None);
+						}
+
+						future::ready(Some(headers))
+					});
+
+					// Flatten the Vec<PubSubResult> into individual PubSubResult items
+					let flat_stream = stream.flat_map(futures::stream::iter);
+
 					PendingSubscription::from(pending)
-						.pipe_from_stream(stream, BoundedVecDeque::new(16))
+						.pipe_from_stream(flat_stream, BoundedVecDeque::new(16))
 						.await
 				}
 				Kind::Logs => {