[pallet-revive] eth-rpc: fix latest block initialization and reduce per-block RPC calls.

.github/workflows/tests-evm.yml

@@ -23,7 +23,7 @@ jobs:
     needs: [preflight]
     runs-on: ${{ needs.preflight.outputs.RUNNER }}
     if: ${{ needs.preflight.outputs.changes_rust }}
-    timeout-minutes: 60
+    timeout-minutes: 120
     container:
       image: ${{ needs.preflight.outputs.IMAGE }}
     permissions:

prdoc/pr_12742.prdoc

@@ -0,0 +1,17 @@
+title: '[pallet-revive] eth-rpc: fix `latest` block initialization and reduce per-block
+  RPC calls.'
+doc:
+- audience: Runtime Dev
+  description: |-
+    Both the finalized and best blocks were initialized using `at_current_block()`, which in subxt 0.50 returns the *finalized* block; the best block therefore started behind the chain tip and served stale blocks until the subscription caught up. Also reduce the number of RPC calls by reusing the already resolved block in `fetch_receipt_data`, and by only requesting receipt data when a block contains EVM extrinsics.
+
+    ## Changes
+
+    - Initialize `latest_block` from the actual best block.
+    - Ignore the blocks that the subscriptions replay when they (re)initialise, instead of walking a tracked head backwards.
+    - Pass the already resolved block to `fetch_receipt_data` instead of a hash.
+    - Only request receipt data when the block contains pallet-revive extrinsics, or when an extrinsic decode fails and its type is yet unknown.
+    - Raise the `differential-tests` job timeout from 60 to 120 minutes; the full corpus no longer finishes within 60 minutes since the subxt 0.50 bump. Further investigation of the slowdown will be tackled separately.
+crates:
+- name: pallet-revive-eth-rpc
+  bump: patch

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

@@ -32,7 +32,8 @@ use tokio::sync::RwLock;
 /// BlockInfoProvider cache and retrieves information about blocks.
 #[async_trait]
 pub trait BlockInfoProvider: Send + Sync {
-	/// Update the latest block
+	/// Update the latest block for the provided `subscription_type` to the given block, which is
+	/// ignored if it is not a valid new head.
 	async fn update_latest(&self, block: Arc<SubstrateBlock>, subscription_type: SubscriptionType);
 
 	/// Return the latest finalized block.
@@ -77,24 +78,41 @@ impl SubxtBlockInfoProvider {
 		api: OnlineClient<SrcChainConfig>,
 		rpc: LegacyRpcMethods<RpcConfigFor<SrcChainConfig>>,
 	) -> Result<Self, ClientError> {
-		let latest = Arc::new(api.at_current_block().await?);
+		let latest_finalized_block = Arc::new(api.at_current_block().await?);
+		let best_hash = rpc.chain_get_block_hash(None).await?.ok_or(ClientError::BlockNotFound)?;
+		let latest_block = Arc::new(api.at_block(best_hash).await?);
 		Ok(Self {
 			api,
 			rpc,
-			latest_block: Arc::new(RwLock::new(latest.clone())),
-			latest_finalized_block: Arc::new(RwLock::new(latest)),
+			latest_block: Arc::new(RwLock::new(latest_block)),
+			latest_finalized_block: Arc::new(RwLock::new(latest_finalized_block)),
 		})
 	}
 }
 
 #[async_trait]
 impl BlockInfoProvider for SubxtBlockInfoProvider {
 	async fn update_latest(&self, block: Arc<SubstrateBlock>, subscription_type: SubscriptionType) {
-		let mut latest = match subscription_type {
-			SubscriptionType::FinalizedBlocks => self.latest_finalized_block.write().await,
-			SubscriptionType::BestBlocks => self.latest_block.write().await,
-		};
-		*latest = block;
+		// Both streams are seeded with the finalized block on subscription (re)init; don't move
+		// the best/finalized back in this scenario.
+		match subscription_type {
+			SubscriptionType::FinalizedBlocks => {
+				let mut latest = self.latest_finalized_block.write().await;
+				if block.block_number() >= latest.block_number() {
+					*latest = block;
+				}
+			},
+			SubscriptionType::BestBlocks => {
+				// Above the finalized height, so treat it as a reorg rather than a replay.
+				let finalized_number = self.latest_finalized_block.read().await.block_number();
+				let mut latest = self.latest_block.write().await;
+				if block.block_number() >= latest.block_number() ||
+					block.block_number() > finalized_number
+				{
+					*latest = block;
+				}
+			},
+		}
 	}
 
 	async fn latest_block(&self) -> Arc<SubstrateBlock> {

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

@@ -735,6 +735,15 @@ impl VersionAwareRuntimeApiProvider {
 		Ok(VersionAwareRuntimeApi::new(at_block, capabilities))
 	}
 
+	/// Returns the version-aware runtime API of the given block handle.
+	pub async fn at_resolved_block(
+		&self,
+		at_block: OnlineClientAtBlock<SrcChainConfig>,
+	) -> Result<VersionAwareRuntimeApi, ClientError> {
+		let capabilities = self.capabilities(&at_block).await?;
+		Ok(VersionAwareRuntimeApi::new(at_block, capabilities))
+	}
+
 	/// Returns the capabilities of the handle's runtime spec version, computing and caching them
 	/// if they are not cached yet.
 	async fn capabilities(

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

@@ -43,7 +43,11 @@ use std::{
 		atomic::{AtomicU64, Ordering},
 	},
 };
-use subxt::events::{DecodeAsEvent, Phase};
+use subxt::{
+	client::OfflineClientAtBlockT,
+	events::{DecodeAsEvent, Phase},
+	extrinsics::Extrinsics,
+};
 
 type EventDetails<'a> = subxt::events::Event<'a, SrcChainConfig>;
 
@@ -153,8 +157,42 @@ fn extract_revive_events(
 	(reverted_extrinsics, logs_by_extrinsic)
 }
 
+/// Returns the revive transactions from a block, and whether any extrinsic decode failed.
+fn extract_eth_transacts<C: OfflineClientAtBlockT<SrcChainConfig>>(
+	block_extrinsics: &Extrinsics<'_, SrcChainConfig, C>,
+	block_number: SubstrateBlockNumber,
+) -> Result<(Vec<(EthTransact, usize)>, bool), ClientError> {
+	let mut extrinsics = Vec::new();
+	let mut undecoded_extrinsic = false;
+	for (ext_idx, ext) in block_extrinsics.iter().enumerate() {
+		let ext = match ext {
+			Ok(ext) => ext,
+			// Don't error here since the call type is unknown.
+			Err(err) => {
+				log::debug!(target: LOG_TARGET,
+					"Failed to decode extrinsic {ext_idx} of block #{block_number}: {err:?}");
+				undecoded_extrinsic = true;
+				continue;
+			},
+		};
+		match ext.decode_call_data_fields_as::<EthTransact>() {
+			Some(Ok(call)) => extrinsics.push((call, ext_idx)),
+			Some(Err(err)) => {
+				log::error!(target: LOG_TARGET,
+					"Failed to decode the EthTransact call in extrinsic {ext_idx} of block \
+					#{block_number}: {err:?}");
+				return Err(subxt::Error::from(err).into());
+			},
+			// Not a revive transaction.
+			None => {},
+		}
+	}
+
+	Ok((extrinsics, undecoded_extrinsic))
+}
+
 type FetchReceiptDataFn = Arc<
-	dyn Fn(H256) -> Pin<Box<dyn Future<Output = Option<Vec<ReceiptGasInfoV1>>> + Send>>
+	dyn Fn(SubstrateBlock) -> Pin<Box<dyn Future<Output = Option<Vec<ReceiptGasInfoV1>>> + Send>>
 		+ Send
 		+ Sync,
 >;
@@ -239,12 +277,13 @@ impl ReceiptExtractor {
 		});
 
 		let provider = runtime_api_provider;
-		let fetch_receipt_data = Arc::new(move |block_hash| {
+		let fetch_receipt_data = Arc::new(move |at_block: SubstrateBlock| {
 			let provider = provider.clone();
 
 			let fut = async move {
+				let block_hash = at_block.block_hash();
 				let runtime_api = provider
-					.at(block_hash)
+					.at_resolved_block(at_block)
 					.await
 					.inspect_err(|err| {
 						log::debug!(
@@ -471,36 +510,25 @@ impl ReceiptExtractor {
 		&self,
 		block: &SubstrateBlock,
 	) -> Result<impl Iterator<Item = (EthTransact, ReceiptGasInfoV1, usize)>, ClientError> {
-		// Filter extrinsics from pallet_revive
-		let extrinsics = block.extrinsics().fetch().await.inspect_err(|err| {
+		let block_extrinsics = block.extrinsics().fetch().await.inspect_err(|err| {
 			log::debug!(target: LOG_TARGET, "Error fetching for #{:?} extrinsics: {err:?}", block.block_number());
 		})?;
 
-		let receipt_data =
-			(self.fetch_receipt_data)(block.block_hash()).await.ok_or_else(|| {
+		let block_number = block.block_number();
+		let (extrinsics, undecoded_extrinsic) =
+			extract_eth_transacts(&block_extrinsics, block_number)?;
+
+		// Skip the runtime query for blocks with no revive extrinsics.
+		let receipt_data = if extrinsics.is_empty() && !undecoded_extrinsic {
+			Vec::new()
+		} else {
+			(self.fetch_receipt_data)(block.clone()).await.ok_or_else(|| {
 				log::trace!(target: LOG_TARGET,
 				"Receipt data not found for block #{} ({:?})",
 				block.block_number(), block.block_hash());
 				ClientError::ReceiptDataNotFound
-			})?;
-		let block_number = block.block_number();
-		let extrinsics: Vec<_> = extrinsics
-			.iter()
-			.enumerate()
-			.flat_map(|(ext_idx, ext)| {
-				let ext = ext.ok()?;
-				match ext.decode_call_data_fields_as::<EthTransact>()? {
-					Ok(call) => Some((call, ext_idx)),
-					Err(err) => {
-						log::warn!(
-							target: LOG_TARGET,
-							"Failed to decode EthTransact call in extrinsic {ext_idx} of block #{block_number}: {err:?}, transaction dropped from receipts"
-						);
-						None
-					},
-				}
-			})
-			.collect();
+			})?
+		};
 
 		// Sanity check we received enough data from the pallet revive.
 		if receipt_data.len() != extrinsics.len() {
@@ -764,19 +792,30 @@ mod tests {
 		metadata::Metadata,
 	};
 
+	/// An offline client carrying the generated runtime metadata for every block.
+	fn offline_client() -> OfflineClient<PolkadotConfig> {
+		let metadata_bytes: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/revive_chain.scale"));
+		let metadata = Metadata::decode(&mut &metadata_bytes[..]).unwrap();
+		let config = PolkadotConfigBuilder::new()
+			.set_metadata_for_spec_versions(std::iter::once((0u32, metadata.into())))
+			.set_spec_version_for_block_ranges(std::iter::once(SpecVersionForRange {
+				block_range: 0..u64::MAX,
+				spec_version: 0,
+				transaction_version: 0,
+			}))
+			.build();
+		OfflineClient::<PolkadotConfig>::new_with_config(config)
+	}
+
 	/// Build `Events` by SCALE-encoding revive events against the generated runtime metadata.
 	struct EventsBuilder {
-		metadata: Metadata,
 		bytes: Vec<u8>,
 		count: u32,
 	}
 
 	impl EventsBuilder {
 		fn new() -> Self {
-			let metadata_bytes: &[u8] =
-				include_bytes!(concat!(env!("OUT_DIR"), "/revive_chain.scale"));
-			let metadata = Metadata::decode(&mut &metadata_bytes[..]).unwrap();
-			Self { metadata, bytes: Vec::new(), count: 0 }
+			Self { bytes: Vec::new(), count: 0 }
 		}
 
 		fn push_event(
@@ -799,15 +838,7 @@ mod tests {
 			Compact(self.count).encode_to(&mut encoded_events);
 			encoded_events.extend(self.bytes);
 
-			let config = PolkadotConfigBuilder::new()
-				.set_metadata_for_spec_versions(std::iter::once((0u32, self.metadata.into())))
-				.set_spec_version_for_block_ranges(std::iter::once(SpecVersionForRange {
-					block_range: 0..u64::MAX,
-					spec_version: 0,
-					transaction_version: 0,
-				}))
-				.build();
-			let client = OfflineClient::<PolkadotConfig>::new_with_config(config);
+			let client = offline_client();
 			let at_block =
 				client.at_block(0u64).expect("spec version range covers all block numbers; qed");
 			at_block.events().from_bytes(encoded_events)
@@ -922,4 +953,65 @@ mod tests {
 		assert_eq!(logs[&0][1].log_index, U256::from(1));
 		assert_eq!(logs[&2][0].log_index, U256::from(3));
 	}
+
+	/// SCALE-encode a bare extrinsic the way a block body carries it, length prefix included.
+	fn encode_bare(call: revive_dev_runtime::RuntimeCall) -> Vec<u8> {
+		let extrinsic: revive_dev_runtime::UncheckedExtrinsic =
+			pallet_revive::evm::runtime::UncheckedExtrinsic(
+				sp_runtime::generic::UncheckedExtrinsic::new_bare(call),
+			);
+		extrinsic.encode()
+	}
+
+	const PAYLOAD: [u8; 4] = [0xde, 0xad, 0xbe, 0xef];
+
+	fn eth_transact_extrinsic() -> Vec<u8> {
+		encode_bare(revive_dev_runtime::RuntimeCall::Revive(pallet_revive::Call::eth_transact {
+			payload: PAYLOAD.to_vec(),
+		}))
+	}
+
+	fn non_revive_extrinsic() -> Vec<u8> {
+		encode_bare(revive_dev_runtime::RuntimeCall::System(frame_system::Call::remark {
+			remark: vec![0x01],
+		}))
+	}
+
+	/// Run the extraction over a synthetic block body.
+	async fn extract_from(
+		blobs: Vec<Vec<u8>>,
+	) -> Result<(Vec<(EthTransact, usize)>, bool), ClientError> {
+		const BLOCK_NUMBER: SubstrateBlockNumber = 42;
+
+		let client = offline_client();
+		let at_block = client
+			.at_block(BLOCK_NUMBER)
+			.expect("spec version range covers every block number; qed");
+		let extrinsics = at_block.extrinsics().from_bytes(blobs).await;
+		extract_eth_transacts(&extrinsics, BLOCK_NUMBER)
+	}
+
+	#[tokio::test]
+	async fn extract_eth_transacts_collects_revive_calls() {
+		let (calls, undecoded) =
+			extract_from(vec![non_revive_extrinsic(), eth_transact_extrinsic()])
+				.await
+				.unwrap();
+
+		assert!(!undecoded, "every extrinsic decoded");
+		assert_eq!(calls.len(), 1, "only the revive extrinsic is collected");
+		assert_eq!(calls[0].1, 1, "the extrinsic index is preserved");
+		assert_eq!(calls[0].0.payload, PAYLOAD, "the call fields are decoded");
+	}
+
+	#[tokio::test]
+	async fn extract_eth_transacts_keeps_revive_calls_next_to_an_undecodable_one() {
+		let (calls, undecoded) =
+			extract_from(vec![vec![0xff; 4], eth_transact_extrinsic()]).await.unwrap();
+
+		assert_eq!(calls.len(), 1, "an undecodable extrinsic must not hide a decoded one");
+		assert_eq!(calls[0].1, 1, "the extrinsic index is preserved");
+		assert_eq!(calls[0].0.payload, PAYLOAD, "the call fields are decoded");
+		assert!(undecoded, "it may be a revive one, so report it");
+	}
 }