Improve block import notification strategy (polkadot-evm#1030)

* Improve block import notification strategy

* oops

* taplo

* clippy

* Notify only when not major syncing

Cargo.toml

@@ -74,6 +74,7 @@ sc-service = { version = "0.10.0-dev", git = "https://github.com/paritytech/subs
 sc-telemetry = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-transaction-pool = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-transaction-pool-api = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-utils = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master" }
 # Substrate Primitive
 sp-api = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 sp-block-builder = { version = "4.0.0-dev", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }

client/mapping-sync/Cargo.toml

@@ -14,13 +14,30 @@ targets = ["x86_64-unknown-linux-gnu"]
 futures = "0.3.25"
 futures-timer = "3.0.1"
 log = "0.4.17"
+parking_lot = "0.12.1"
 # Substrate
 sc-client-api = { workspace = true }
 sp-api = { workspace = true }
 sp-blockchain = { workspace = true }
+sp-consensus = { workspace = true, features = ["default"] }
 sp-runtime = { workspace = true }
 # Frontier
 fc-db = { workspace = true }
 fc-storage = { workspace = true }
 fp-consensus = { workspace = true, features = ["default"] }
 fp-rpc = { workspace = true, features = ["default"] }
+sc-utils = { workspace = true }
+
+[dev-dependencies]
+ethereum = { workspace = true, features = ["with-codec"] }
+ethereum-types = { workspace = true }
+tempfile = "3.3.0"
+tokio = { version = "1.24", features = ["sync"] }
+#Frontier
+fp-storage = { workspace = true, features = ["default"] }
+frontier-template-runtime = { workspace = true, features = ["default"] }
+# Substrate
+sc-block-builder = { workspace = true }
+sc-client-db = { workspace = true }
+sp-core = { workspace = true, features = ["default"] }
+substrate-test-runtime-client = { workspace = true }

client/mapping-sync/src/lib.rs

@@ -29,12 +29,22 @@ use std::sync::Arc;
 use sc_client_api::backend::{Backend, StorageProvider};
 use sp_api::{ApiExt, ProvideRuntimeApi};
 use sp_blockchain::{Backend as _, HeaderBackend};
+use sp_consensus::SyncOracle;
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT, Zero};
 // Frontier
 use fc_storage::OverrideHandle;
 use fp_consensus::{FindLogError, Hashes, Log, PostLog, PreLog};
 use fp_rpc::EthereumRuntimeRPCApi;
 
+pub type EthereumBlockNotificationSinks<T> =
+	parking_lot::Mutex<Vec<sc_utils::mpsc::TracingUnboundedSender<T>>>;
+
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub struct EthereumBlockNotification<Block: BlockT> {
+	pub is_new_best: bool,
+	pub hash: Block::Hash,
+}
+
 pub fn sync_block<Block: BlockT, C, BE>(
 	client: &C,
 	overrides: Arc<OverrideHandle<Block>>,
@@ -160,6 +170,10 @@ pub fn sync_one_block<Block: BlockT, C, BE>(
 	frontier_backend: &fc_db::Backend<Block>,
 	sync_from: <Block::Header as HeaderT>::Number,
 	strategy: SyncStrategy,
+	sync_oracle: Arc<dyn SyncOracle + Send + Sync + 'static>,
+	pubsub_notification_sinks: Arc<
+		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
+	>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -208,7 +222,6 @@ where
 		frontier_backend
 			.meta()
 			.write_current_syncing_tips(current_syncing_tips)?;
-		Ok(true)
 	} else {
 		if SyncStrategy::Parachain == strategy
 			&& operating_header.number() > &client.info().best_number
@@ -221,8 +234,22 @@ where
 		frontier_backend
 			.meta()
 			.write_current_syncing_tips(current_syncing_tips)?;
-		Ok(true)
 	}
+	// Notify on import and remove closed channels.
+	// Only notify when the node is node 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()
+		} else {
+			// Remove from the pool if in major syncing.
+			false
+		}
+	});
+	Ok(true)
 }
 
 pub fn sync_blocks<Block: BlockT, C, BE>(
@@ -233,6 +260,10 @@ pub fn sync_blocks<Block: BlockT, C, BE>(
 	limit: usize,
 	sync_from: <Block::Header as HeaderT>::Number,
 	strategy: SyncStrategy,
+	sync_oracle: Arc<dyn SyncOracle + Send + Sync + 'static>,
+	pubsub_notification_sinks: Arc<
+		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
+	>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -251,6 +282,8 @@ where
 				frontier_backend,
 				sync_from,
 				strategy,
+				sync_oracle.clone(),
+				pubsub_notification_sinks.clone(),
 			)?;
 	}