fix: 🐛 extend EthereumBlockNotification with reorg info to allow compliance with Ethereum specs

Author: manuelmauro

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

@@ -22,7 +22,7 @@ mod worker;
 
 pub use worker::MappingSyncWorker;
 
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 // Substrate
 use sc_client_api::backend::{Backend, StorageProvider};
@@ -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,6 +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, BestBlockInfo<Block>>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -223,14 +225,24 @@ where
 			.write_current_syncing_tips(current_syncing_tips)?;
 	}
 	// Notify on import and remove closed channels.
-	// Only notify when the node is node in major syncing.
+	// Only notify when the node is not in major syncing.
 	let sinks = &mut pubsub_notification_sinks.lock();
 	sinks.retain(|sink| {
 		if !sync_oracle.is_major_syncing() {
 			let hash = operating_header.hash();
-			let is_new_best = client.info().best_hash == hash;
-			sink.unbounded_send(EthereumBlockNotification { is_new_best, hash })
-				.is_ok()
+			// Use the `is_new_best` status from import time if available.
+			// 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 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
@@ -251,6 +263,7 @@ pub fn sync_blocks<Block: BlockT, C, BE>(
 	pubsub_notification_sinks: Arc<
 		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
 	>,
+	best_at_import: &mut HashMap<Block::Hash, BestBlockInfo<Block>>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -271,9 +284,16 @@ where
 				strategy,
 				sync_oracle.clone(),
 				pubsub_notification_sinks.clone(),
+				best_at_import,
 			)?;
 	}
 
+	// Prune old entries from best_at_import to prevent unbounded growth.
+	// 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(|_, info| info.block_number > finalized_number);
+
 	Ok(synced_any)
 }
 

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

@@ -16,7 +16,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use std::{pin::Pin, sync::Arc, time::Duration};
+use std::{collections::HashMap, pin::Pin, sync::Arc, time::Duration};
 
 use futures::{
 	prelude::*,
@@ -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>,
@@ -57,6 +65,13 @@ pub struct MappingSyncWorker<Block: BlockT, C, BE> {
 	sync_oracle: Arc<dyn SyncOracle + Send + Sync + 'static>,
 	pubsub_notification_sinks:
 		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 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, BestBlockInfo<Block>>,
 }
 
 impl<Block: BlockT, C, BE> Unpin for MappingSyncWorker<Block, C, BE> {}
@@ -94,6 +109,7 @@ impl<Block: BlockT, C, BE> MappingSyncWorker<Block, C, BE> {
 
 			sync_oracle,
 			pubsub_notification_sinks,
+			best_at_import: HashMap::new(),
 		}
 	}
 }
@@ -114,8 +130,42 @@ where
 		loop {
 			match Stream::poll_next(Pin::new(&mut self.import_notifications), cx) {
 				Poll::Pending => break,
-				Poll::Ready(Some(_)) => {
+				Poll::Ready(Some(notification)) => {
 					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,
+					// and reorg info if this block became best as part of a reorg.
+					if notification.is_new_best {
+						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();
+							// tree_route.enacted() returns blocks from common ancestor (exclusive)
+							// to new best (EXCLUSIVE). We need to include the new best block
+							// (notification.hash) to get the complete enacted list.
+							let mut enacted: Vec<_> = tree_route
+								.enacted()
+								.iter()
+								.map(|hash_and_number| hash_and_number.hash)
+								.collect();
+							enacted.push(notification.hash);
+							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),
 			}
@@ -138,7 +188,12 @@ where
 		if fire {
 			self.inner_delay = None;
 
-			match crate::kv::sync_blocks(
+			// Temporarily take ownership of best_at_import to avoid borrow checker issues
+			// (we can't have both an immutable borrow of self.client and a mutable borrow
+			// of self.best_at_import at the same time)
+			let mut best_at_import = std::mem::take(&mut self.best_at_import);
+
+			let result = crate::kv::sync_blocks(
 				self.client.as_ref(),
 				self.substrate_backend.as_ref(),
 				self.storage_override.clone(),
@@ -148,7 +203,13 @@ where
 				self.strategy,
 				self.sync_oracle.clone(),
 				self.pubsub_notification_sinks.clone(),
-			) {
+				&mut best_at_import,
+			);
+
+			// Restore the best_at_import set
+			self.best_at_import = best_at_import;
+
+			match result {
 				Ok(have_next) => {
 					self.have_next = have_next;
 					Poll::Ready(Some(()))

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,7 +29,15 @@ 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};
+
+/// Reorg information passed between commands.
+#[derive(Debug, Clone)]
+pub struct ReorgData {
+	pub common_ancestor: H256,
+	pub enacted: Vec<H256>,
+	pub retracted: Vec<H256>,
+}
 
 /// Defines the commands for the sync worker.
 #[derive(Debug)]
@@ -39,7 +47,12 @@ pub enum WorkerCommand {
 	/// Index leaves.
 	IndexLeaves(Vec<H256>),
 	/// Index the best block known so far via import notifications.
-	IndexBestBlock(H256),
+	/// When `reorg_data` is Some, emits a pubsub notification with reorg info after indexing.
+	/// When `reorg_data` is None, emits a simple new block notification.
+	IndexBestBlock {
+		block_hash: H256,
+		reorg_data: Option<ReorgData>,
+	},
 	/// Canonicalize the enacted and retracted blocks reported via import notifications.
 	Canonicalize {
 		common: H256,
@@ -122,19 +135,28 @@ where
 							.await;
 						}
 					}
-					WorkerCommand::IndexBestBlock(block_hash) => {
+					WorkerCommand::IndexBestBlock {
+						block_hash,
+						reorg_data,
+					} => {
 						index_canonical_block_and_ancestors(
 							client.clone(),
 							substrate_backend.clone(),
 							indexer_backend.clone(),
 							block_hash,
 						)
 						.await;
+						// Emit notification after indexing so blocks are available via storage_override
 						let sinks = &mut pubsub_notification_sinks.lock();
 						for sink in sinks.iter() {
 							let _ = sink.unbounded_send(EthereumBlockNotification {
 								is_new_best: true,
 								hash: block_hash,
+								reorg_info: reorg_data.as_ref().map(|data| ReorgInfo {
+									common_ancestor: data.common_ancestor,
+									retracted: data.retracted.clone(),
+									enacted: data.enacted.clone(),
+								}),
 							});
 						}
 					}
@@ -145,6 +167,7 @@ where
 					} => {
 						canonicalize_blocks(indexer_backend.clone(), common, enacted, retracted)
 							.await;
+						// Notification is emitted by IndexBestBlock after indexing completes
 					}
 					WorkerCommand::CheckIndexedBlocks => {
 						// Fix any indexed blocks that did not have their logs indexed
@@ -229,7 +252,7 @@ where
 						notification.is_new_best,
 					);
 					if notification.is_new_best {
-						if let Some(tree_route) = notification.tree_route {
+						let reorg_data = if let Some(ref tree_route) = notification.tree_route {
 							log::debug!(
 								target: "frontier-sql",
 								"🔀  Re-org happened at new best {}, proceeding to canonicalize db",
@@ -240,21 +263,38 @@ where
 								.iter()
 								.map(|hash_and_number| hash_and_number.hash)
 								.collect::<Vec<_>>();
-							let enacted = tree_route
+							// tree_route.enacted() returns blocks from common ancestor (exclusive)
+							// to new best (exclusive). We need to include the new best block
+							// (notification.hash) to get the complete enacted list.
+							let mut enacted: Vec<_> = tree_route
 								.enacted()
 								.iter()
 								.map(|hash_and_number| hash_and_number.hash)
-								.collect::<Vec<_>>();
+								.collect();
+							enacted.push(notification.hash);
 
 							let common = tree_route.common_block().hash;
 							tx.send(WorkerCommand::Canonicalize {
 								common,
+								enacted: enacted.clone(),
+								retracted: retracted.clone(),
+							}).await.ok();
+							Some(ReorgData {
+								common_ancestor: common,
 								enacted,
 								retracted,
-							}).await.ok();
-						}
+							})
+						} else {
+							None
+						};
 
-						tx.send(WorkerCommand::IndexBestBlock(notification.hash)).await.ok();
+						// Index the best block and emit notification after indexing completes.
+						// This ensures blocks are available via storage_override when the
+						// notification is processed.
+						tx.send(WorkerCommand::IndexBestBlock {
+							block_hash: notification.hash,
+							reorg_data,
+						}).await.ok();
 					}
 				}
 			}