Improve block import notification strategy (#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(),
  )?;
  }