Improve block import notification strategy

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,6 +14,7 @@ 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 }
@@ -24,3 +25,19 @@ 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 }
+sp-core = { workspace = true, features = ["default"] }
+sp-consensus = { workspace = true, features = ["default"] }
+sc-client-db = { workspace = true }
+substrate-test-runtime-client = { workspace = true }

client/mapping-sync/src/lib.rs

@@ -35,6 +35,15 @@ 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 +169,9 @@ pub fn sync_one_block<Block: BlockT, C, BE>(
 	frontier_backend: &fc_db::Backend<Block>,
 	sync_from: <Block::Header as HeaderT>::Number,
 	strategy: SyncStrategy,
+	pubsub_notification_sinks: Arc<
+		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
+	>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -208,7 +220,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 +232,17 @@ where
 		frontier_backend
 			.meta()
 			.write_current_syncing_tips(current_syncing_tips)?;
-		Ok(true)
 	}
+	// Notify on import and remove closed channels.
+	let sinks = &mut pubsub_notification_sinks.lock();
+	sinks.retain(|sink| {
+		sink.unbounded_send(EthereumBlockNotification {
+			is_new_best: true,
+			hash: client.info().best_hash,
+		})
+		.is_ok()
+	});
+	Ok(true)
 }
 
 pub fn sync_blocks<Block: BlockT, C, BE>(
@@ -233,6 +253,9 @@ pub fn sync_blocks<Block: BlockT, C, BE>(
 	limit: usize,
 	sync_from: <Block::Header as HeaderT>::Number,
 	strategy: SyncStrategy,
+	pubsub_notification_sinks: Arc<
+		EthereumBlockNotificationSinks<EthereumBlockNotification<Block>>,
+	>,
 ) -> Result<bool, String>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -251,6 +274,7 @@ where
 				frontier_backend,
 				sync_from,
 				strategy,
+				pubsub_notification_sinks.clone(),
 			)?;
 	}
 

client/mapping-sync/src/worker.rs

@@ -56,6 +56,9 @@ pub struct MappingSyncWorker<Block: BlockT, C, BE> {
 	retry_times: usize,
 	sync_from: <Block::Header as HeaderT>::Number,
 	strategy: SyncStrategy,
+
+	pubsub_notification_sinks:
+		Arc<crate::EthereumBlockNotificationSinks<crate::EthereumBlockNotification<Block>>>,
 }
 
 impl<Block: BlockT, C, BE> Unpin for MappingSyncWorker<Block, C, BE> {}
@@ -71,6 +74,9 @@ impl<Block: BlockT, C, BE> MappingSyncWorker<Block, C, BE> {
 		retry_times: usize,
 		sync_from: <Block::Header as HeaderT>::Number,
 		strategy: SyncStrategy,
+		pubsub_notification_sinks: Arc<
+			crate::EthereumBlockNotificationSinks<crate::EthereumBlockNotification<Block>>,
+		>,
 	) -> Self {
 		Self {
 			import_notifications,
@@ -86,6 +92,8 @@ impl<Block: BlockT, C, BE> MappingSyncWorker<Block, C, BE> {
 			retry_times,
 			sync_from,
 			strategy,
+
+			pubsub_notification_sinks,
 		}
 	}
 }
@@ -137,6 +145,7 @@ where
 				self.retry_times,
 				self.sync_from,
 				self.strategy,
+				self.pubsub_notification_sinks.clone(),
 			) {
 				Ok(have_next) => {
 					self.have_next = have_next;
@@ -153,3 +162,190 @@ where
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::{EthereumBlockNotification, EthereumBlockNotificationSinks};
+	use fc_storage::{OverrideHandle, SchemaV3Override, StorageOverride};
+	use fp_storage::{EthereumStorageSchema, PALLET_ETHEREUM_SCHEMA};
+	use futures::executor;
+	use sc_block_builder::BlockBuilderProvider;
+	use sc_client_api::BlockchainEvents;
+	use sp_api::Encode;
+	use sp_consensus::BlockOrigin;
+	use sp_core::{H160, H256, U256};
+	use sp_runtime::{generic::Header, traits::BlakeTwo256, Digest};
+	use std::collections::BTreeMap;
+	use substrate_test_runtime_client::{
+		ClientBlockImportExt, DefaultTestClientBuilderExt, TestClientBuilder, TestClientBuilderExt,
+	};
+	use tempfile::tempdir;
+
+	type OpaqueBlock = sp_runtime::generic::Block<
+		Header<u64, BlakeTwo256>,
+		substrate_test_runtime_client::runtime::Extrinsic,
+	>;
+
+	fn ethereum_digest() -> Digest {
+		let partial_header = ethereum::PartialHeader {
+			parent_hash: H256::random(),
+			beneficiary: H160::default(),
+			state_root: H256::default(),
+			receipts_root: H256::default(),
+			logs_bloom: ethereum_types::Bloom::default(),
+			difficulty: U256::zero(),
+			number: U256::zero(),
+			gas_limit: U256::zero(),
+			gas_used: U256::zero(),
+			timestamp: 0u64,
+			extra_data: Vec::new(),
+			mix_hash: H256::default(),
+			nonce: ethereum_types::H64::default(),
+		};
+		let ethereum_block = ethereum::Block::new(partial_header, vec![], vec![]);
+		Digest {
+			logs: vec![sp_runtime::generic::DigestItem::Consensus(
+				fp_consensus::FRONTIER_ENGINE_ID,
+				fp_consensus::PostLog::Hashes(fp_consensus::Hashes::from_block(ethereum_block))
+					.encode(),
+			)],
+		}
+	}
+
+	#[tokio::test]
+	async fn block_import_notification_works() {
+		let tmp = tempdir().expect("create a temporary directory");
+		let builder = TestClientBuilder::new().add_extra_storage(
+			PALLET_ETHEREUM_SCHEMA.to_vec(),
+			Encode::encode(&EthereumStorageSchema::V3),
+		);
+		// Backend
+		let backend = builder.backend();
+		// Client
+		let (client, _) =
+			builder.build_with_native_executor::<frontier_template_runtime::RuntimeApi, _>(None);
+		let mut client = Arc::new(client);
+		// Overrides
+		let mut overrides_map = BTreeMap::new();
+		overrides_map.insert(
+			EthereumStorageSchema::V3,
+			Box::new(SchemaV3Override::new(client.clone())) as Box<dyn StorageOverride<_>>,
+		);
+		let overrides = Arc::new(OverrideHandle {
+			schemas: overrides_map,
+			fallback: Box::new(SchemaV3Override::new(client.clone())),
+		});
+
+		let frontier_backend = Arc::new(
+			fc_db::Backend::<OpaqueBlock>::new(
+				client.clone(),
+				&fc_db::DatabaseSettings {
+					source: sc_client_db::DatabaseSource::RocksDb {
+						path: tmp.path().to_path_buf(),
+						cache_size: 0,
+					},
+				},
+			)
+			.expect("frontier backend"),
+		);
+
+		let notification_stream = client.clone().import_notification_stream();
+		let client_inner = client.clone();
+
+		let pubsub_notification_sinks: EthereumBlockNotificationSinks<
+			EthereumBlockNotification<OpaqueBlock>,
+		> = Default::default();
+		let pubsub_notification_sinks = Arc::new(pubsub_notification_sinks);
+
+		let pubsub_notification_sinks_inner = pubsub_notification_sinks.clone();
+
+		tokio::task::spawn(async move {
+			MappingSyncWorker::new(
+				notification_stream,
+				Duration::new(6, 0),
+				client_inner,
+				backend,
+				overrides.clone(),
+				frontier_backend,
+				3,
+				0,
+				SyncStrategy::Normal,
+				pubsub_notification_sinks_inner,
+			)
+			.for_each(|()| future::ready(()))
+			.await
+		});
+
+		{
+			// A new mpsc channel
+			let (inner_sink, mut block_notification_stream) =
+				sc_utils::mpsc::tracing_unbounded("pubsub_notification_stream", 100_000);
+
+			{
+				// This scope represents a call to eth_subscribe, where it briefly locks the pool
+				// to push the new sink.
+				let sinks = &mut pubsub_notification_sinks.lock();
+				// Push to sink pool
+				sinks.push(inner_sink);
+			}
+
+			// Let's produce a block, which we expect to trigger a channel message
+			let mut builder = client.new_block(ethereum_digest()).unwrap();
+			let block = builder.build().unwrap().block;
+			let block_hash = block.header.hash();
+			client.import(BlockOrigin::Own, block).await;
+
+			// Receive
+			assert_eq!(
+				block_notification_stream
+					.next()
+					.await
+					.expect("a message")
+					.hash,
+				block_hash
+			);
+		}
+
+		{
+			// Assert we still hold a sink in the pool after switching scopes
+			let sinks = pubsub_notification_sinks.lock();
+			assert_eq!(sinks.len(), 1);
+		}
+
+		{
+			// Create yet another mpsc channel
+			let (inner_sink, mut block_notification_stream) =
+				sc_utils::mpsc::tracing_unbounded("pubsub_notification_stream", 100_000);
+
+			{
+				let sinks = &mut pubsub_notification_sinks.lock();
+				// Push it
+				sinks.push(inner_sink);
+				// Now we expect two sinks in the pool
+				assert_eq!(sinks.len(), 2);
+			}
+
+			// Let's produce another block, this not only triggers a message in the new channel
+			// but also removes the closed channels from the pool.
+			let mut builder = client.new_block(ethereum_digest()).unwrap();
+			let block = builder.build().unwrap().block;
+			let block_hash = block.header.hash();
+			client.import(BlockOrigin::Own, block).await;
+
+			// Receive
+			assert_eq!(
+				block_notification_stream
+					.next()
+					.await
+					.expect("a message")
+					.hash,
+				block_hash
+			);
+
+			// So we expect the pool to hold one sink only after cleanup
+			let sinks = &mut pubsub_notification_sinks.lock();
+			assert_eq!(sinks.len(), 1);
+		}
+	}
+}