Implement eth_subscribe

prdoc/pr_11081.prdoc

@@ -0,0 +1,12 @@
+title: Implement `eth_subscribe`
+doc:
+- audience: Runtime Dev
+  description: |-
+    # Description
+
+    Implemented `eth_subscribe` in the eth-rpc. The subscription kinds implemented is `newHeads` and `logs`.
+crates:
+- name: pallet-revive-eth-rpc
+  bump: minor
+- name: pallet-revive
+  bump: minor

substrate/frame/revive/rpc/Cargo.toml

@@ -35,7 +35,10 @@ sc-cli = { workspace = true, default-features = true }
 sc-rpc = { workspace = true, default-features = true }
 sc-rpc-api = { workspace = true, default-features = true }
 sc-service = { workspace = true, default-features = true }
-serde = { workspace = true, default-features = true, features = ["alloc", "derive"] }
+serde = { workspace = true, default-features = true, features = [
+	"alloc",
+	"derive",
+] }
 serde_json = { workspace = true }
 sp-arithmetic = { workspace = true, default-features = true }
 sp-core = { workspace = true, default-features = true }
@@ -45,10 +48,13 @@ sp-runtime = { workspace = true, default-features = true }
 sp-timestamp = { workspace = true }
 sp-weights = { workspace = true, default-features = true }
 sqlx = { workspace = true, features = ["macros", "runtime-tokio", "sqlite"] }
-subxt = { workspace = true, default-features = true, features = ["reconnecting-rpc-client"] }
+subxt = { workspace = true, default-features = true, features = [
+	"reconnecting-rpc-client",
+] }
 subxt-signer = { workspace = true, features = ["unstable-eth"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+tokio-stream = { workspace = true }
 
 [dev-dependencies]
 env_logger = { workspace = true }

substrate/frame/revive/rpc/src/apis/execution_apis.rs

@@ -186,4 +186,14 @@ pub trait EthRpc {
 		newest_block: BlockNumberOrTag,
 		reward_percentiles: Option<Vec<f64>>,
 	) -> RpcResult<FeeHistoryResult>;
+
+	/// Creates a subscription to specific events, returning a subscription ID.
+	/// Notifications are sent for each event matching the subscription via
+	/// `eth_subscription`.
+	#[subscription(
+		name = "eth_subscribe" => "eth_subscription",
+		unsubscribe = "eth_unsubscribe",
+		item = SubscriptionItem
+	)]
+	async fn eth_subscribe(&self, kind: SubscriptionKind, options: Option<SubscriptionOptions>);
 }

substrate/frame/revive/rpc/src/client.rs

@@ -249,6 +249,11 @@ pub struct Client {
 	block_notifier: Option<tokio::sync::broadcast::Sender<H256>>,
 	/// A lock to ensure only one subscription can perform write operations at a time.
 	subscription_lock: Arc<Mutex<()>>,
+
+	/// Block subscription sender side.
+	block_subscription_tx: tokio::sync::broadcast::Sender<Block>,
+	/// Log subscription sender side.
+	log_subscription_tx: tokio::sync::broadcast::Sender<Log>,
 	/// Whether archive mode is enabled
 	is_archive: bool,
 	/// Whether historic backfill has completed. `false` if not started or in progress.
@@ -331,6 +336,8 @@ impl Client {
 			block_notifier: automine
 				.then(|| tokio::sync::broadcast::channel::<H256>(NOTIFIER_CAPACITY).0),
 			subscription_lock: Arc::new(Mutex::new(())),
+			block_subscription_tx: tokio::sync::broadcast::channel(256).0,
+			log_subscription_tx: tokio::sync::broadcast::channel(1000).0,
 			is_archive,
 			backfill_complete: Arc::new(AtomicBool::new(false)),
 		};
@@ -464,6 +471,24 @@ impl Client {
 				},
 				_ => {},
 			}
+
+			// Broadcast the best blocks
+			if let SubscriptionType::BestBlocks = subscription_type &&
+				self.block_subscription_tx.receiver_count() > 0
+			{
+				let _ = self.block_subscription_tx.send(evm_block);
+			}
+
+			// Broadcast the logs, we require a finalized subscription for this so that all of the
+			// events we broadcast are finalized and not prone to reorgs.
+			if let SubscriptionType::FinalizedBlocks = subscription_type &&
+				self.log_subscription_tx.receiver_count() > 0
+			{
+				receipts.iter().flat_map(|receipt| receipt.logs.iter()).for_each(|log| {
+					let _ = self.log_subscription_tx.send(log.clone());
+				});
+			}
+
 			Ok(())
 		})
 		.await
@@ -920,6 +945,16 @@ impl Client {
 	pub async fn get_automine(&self) -> bool {
 		get_automine(&self.rpc_client).await
 	}
+
+	/// Gets the block subscription rx side of the channel.
+	pub fn get_block_subscription_rx(&self) -> tokio::sync::broadcast::Receiver<Block> {
+		self.block_subscription_tx.subscribe()
+	}
+
+	/// Gets the log subscription rx side of the channel.
+	pub fn get_log_subscription_rx(&self) -> tokio::sync::broadcast::Receiver<Log> {
+		self.log_subscription_tx.subscribe()
+	}
 }
 
 fn to_hex(bytes: impl AsRef<[u8]>) -> String {

substrate/frame/revive/rpc/src/lib.rs

@@ -18,14 +18,18 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
 use client::ClientError;
+use futures::{Stream, StreamExt, TryStreamExt};
 use jsonrpsee::{
+	PendingSubscriptionSink, SubscriptionMessage, SubscriptionSink,
 	core::{RpcResult, async_trait},
 	types::{ErrorCode, ErrorObjectOwned},
 };
 use pallet_revive::evm::*;
 use sp_core::{H160, H256, U256, keccak_256};
 use subxt::backend::legacy::rpc_methods::TransactionStatus;
+use subxt_signer::bip39::core::pin::Pin;
 use thiserror::Error;
+use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
 
 mod block_sync;
 pub(crate) use block_sync::{ChainMetadata, SyncLabel, SyncStateKey};
@@ -492,6 +496,41 @@ impl EthRpcServer for EthRpcServerImpl {
 		let result = self.client.fee_history(block_count, newest_block, reward_percentiles).await?;
 		Ok(result)
 	}
+
+	async fn eth_subscribe(
+		&self,
+		pending: PendingSubscriptionSink,
+		kind: SubscriptionKind,
+		options: Option<SubscriptionOptions>,
+	) {
+		let Some(subscription_parameters) = SubscriptionParameters::new(kind, options) else {
+			return pending
+				.reject(ErrorObjectOwned::owned(
+					jsonrpsee::types::error::INVALID_PARAMS_CODE,
+					"Invalid subscription parameters",
+					None::<()>,
+				))
+				.await;
+		};
+		let Ok(sink) = pending.accept().await else {
+			return;
+		};
+
+		let stream: Pin<
+			Box<dyn Stream<Item = Result<SubscriptionItem, BroadcastStreamRecvError>> + Send>,
+		> = match subscription_parameters {
+			SubscriptionParameters::NewBlockHeaders => Box::pin(
+				BroadcastStream::new(self.client.get_block_subscription_rx())
+					.map_ok(|block| SubscriptionItem::BlockHeader(BlockHeader::from(block))),
+			) as _,
+			SubscriptionParameters::Logs(filter) => Box::pin(
+				BroadcastStream::new(self.client.get_log_subscription_rx())
+					.try_filter(move |log| futures::future::ready(filter.matches(log)))
+					.map_ok(SubscriptionItem::Log),
+			) as _,
+		};
+		let _ = tokio::spawn(Self::handle_subscription_forwarding(sink, stream));
+	}
 }
 
 impl EthRpcServerImpl {
@@ -516,4 +555,38 @@ impl EthRpcServerImpl {
 
 		Ok(Some(TransactionInfo::new(&receipt, signed_tx)))
 	}
+
+	async fn handle_subscription_forwarding(
+		sink: SubscriptionSink,
+		mut stream: Pin<
+			Box<dyn Stream<Item = Result<SubscriptionItem, BroadcastStreamRecvError>> + Send>,
+		>,
+	) {
+		loop {
+			tokio::select! {
+				_ = sink.closed() => break,
+				item = stream.next() => {
+					match item {
+						// Stream ended.
+						None => break,
+						// Send the item to the subscriber.
+						Some(Ok(sub_item)) => {
+							let msg = SubscriptionMessage::from_json(&sub_item)
+								.expect("SubscriptionItem is serializable; qed");
+							if sink.send(msg).await.is_err() {
+								break;
+							}
+						},
+						// Broadcast receiver lagged behind — missed messages.
+						Some(Err(BroadcastStreamRecvError::Lagged(count))) => {
+							log::warn!(
+								target: LOG_TARGET,
+								"Subscription lagged, skipped {count} messages"
+							);
+						},
+					}
+				}
+			}
+		}
+	}
 }