From 03251f9810b4dc04a99c5cbd39071d43b523ee3f Mon Sep 17 00:00:00 2001 From: soloking1412 Date: Sat, 18 Jul 2026 23:21:22 +0530 Subject: [PATCH 1/3] pallet-revive: implement the eth_newFilter polling filter API --- Cargo.lock | 1 + prdoc/pr_0000.prdoc | 29 ++ substrate/frame/revive/rpc/Cargo.toml | 1 + .../revive/rpc/src/apis/execution_apis.rs | 23 ++ substrate/frame/revive/rpc/src/filters.rs | 258 ++++++++++++++++++ substrate/frame/revive/rpc/src/lib.rs | 141 +++++++++- substrate/frame/revive/rpc/src/tests.rs | 103 +++++++ 7 files changed, 555 insertions(+), 1 deletion(-) create mode 100644 prdoc/pr_0000.prdoc create mode 100644 substrate/frame/revive/rpc/src/filters.rs diff --git a/Cargo.lock b/Cargo.lock index dda0d4d5356a..58b95415b281 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14117,6 +14117,7 @@ dependencies = [ "pallet-revive-types", "parity-scale-codec", "pretty_assertions", + "rand 0.8.5", "revive-dev-node", "revive-dev-runtime", "rlp 0.6.1", diff --git a/prdoc/pr_0000.prdoc b/prdoc/pr_0000.prdoc new file mode 100644 index 000000000000..3f596ee3625e --- /dev/null +++ b/prdoc/pr_0000.prdoc @@ -0,0 +1,29 @@ +title: 'pallet-revive: implement the eth_newFilter polling filter API' + +doc: +- audience: Node Dev + description: |- + The `pallet-revive` Ethereum JSON-RPC server only offered `eth_subscribe` (a WebSocket push + API) for watching logs and new blocks, leaving the standard HTTP polling filter family + unimplemented. Tooling that cannot hold a WebSocket open — many indexers, serverless + functions, and the default transports of several Ethereum client libraries — relies on that + family and could not watch the chain. + + The following `execution-apis` methods are now supported, matching go-ethereum: + + - `eth_newFilter`: install a log filter (address/topics/range) and get an id. + - `eth_newBlockFilter`: install a filter that reports the hashes of new blocks. + - `eth_getFilterChanges`: poll a filter for the logs or block hashes seen since the last poll. + - `eth_getFilterLogs`: return all logs matching a log filter over its full range. + - `eth_uninstallFilter`: remove a filter. + + Filters are pull-based: each one only remembers the next block to report from, and the log + lookup reuses the existing `eth_getLogs` machinery. Ids are unpredictable (128-bit random) so + a client cannot enumerate or tamper with filters it does not own, and a filter that is not + polled within five minutes is evicted, mirroring go-ethereum and bounding memory use. + `eth_newPendingTransactionFilter` is intentionally out of scope, as the server does not expose + a pending-transaction pool (the same reason `eth_subscribe` has no `newPendingTransactions`). + +crates: +- name: pallet-revive-eth-rpc + bump: minor diff --git a/substrate/frame/revive/rpc/Cargo.toml b/substrate/frame/revive/rpc/Cargo.toml index d427248c77c8..e4102f2e839e 100644 --- a/substrate/frame/revive/rpc/Cargo.toml +++ b/substrate/frame/revive/rpc/Cargo.toml @@ -36,6 +36,7 @@ log = { workspace = true } pallet-revive = { workspace = true, default-features = true } pallet-revive-types = { workspace = true, features = ["std"] } prometheus-endpoint = { workspace = true, default-features = true } +rand = { workspace = true, features = ["std", "std_rng"] } rlp = { workspace = true } sc-cli = { workspace = true, default-features = true } sc-rpc = { workspace = true, default-features = true } diff --git a/substrate/frame/revive/rpc/src/apis/execution_apis.rs b/substrate/frame/revive/rpc/src/apis/execution_apis.rs index b42f421cd738..65c402eda8b5 100644 --- a/substrate/frame/revive/rpc/src/apis/execution_apis.rs +++ b/substrate/frame/revive/rpc/src/apis/execution_apis.rs @@ -99,6 +99,29 @@ pub trait EthRpc { #[method(name = "eth_getLogs")] async fn get_logs(&self, filter: Option) -> RpcResult; + /// Creates a log filter, returning its id. Use `eth_getFilterChanges` to poll it for the logs + /// that matched since the last poll. + #[method(name = "eth_newFilter")] + async fn new_filter(&self, filter: Filter) -> RpcResult; + + /// Creates a block filter, returning its id. Use `eth_getFilterChanges` to poll it for the + /// hashes of blocks added since the last poll. + #[method(name = "eth_newBlockFilter")] + async fn new_block_filter(&self) -> RpcResult; + + /// Polls the filter with the given id, returning the logs (for a log filter) or block hashes + /// (for a block filter) that occurred since the last poll. + #[method(name = "eth_getFilterChanges")] + async fn get_filter_changes(&self, filter_id: U256) -> RpcResult; + + /// Returns all logs matching the log filter with the given id, over its full range. + #[method(name = "eth_getFilterLogs")] + async fn get_filter_logs(&self, filter_id: U256) -> RpcResult; + + /// Uninstalls the filter with the given id. Returns `true` if the filter existed. + #[method(name = "eth_uninstallFilter")] + async fn uninstall_filter(&self, filter_id: U256) -> RpcResult; + /// Returns the value from a storage position at a given address. #[method(name = "eth_getStorageAt")] async fn get_storage_at( diff --git a/substrate/frame/revive/rpc/src/filters.rs b/substrate/frame/revive/rpc/src/filters.rs new file mode 100644 index 000000000000..1ff341d2156b --- /dev/null +++ b/substrate/frame/revive/rpc/src/filters.rs @@ -0,0 +1,258 @@ +// This file is part of Substrate. + +// Copyright (C) Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: Apache-2.0 + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! In-memory registry backing the polling filter API (`eth_newFilter`, `eth_newBlockFilter`, +//! `eth_getFilterChanges`, `eth_getFilterLogs`, `eth_uninstallFilter`). +//! +//! Unlike `eth_subscribe`, these methods are pull-based: a client installs a filter and later +//! polls it for whatever changed since the previous poll. The registry therefore only needs to +//! remember, per filter, the next block to report from; the actual log lookup is served on demand +//! by reusing [`crate::client::Client::logs`]. + +use crate::{client::SubstrateBlockNumber, *}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex, MutexGuard}, + time::{Duration, Instant}, +}; + +/// Filters that are not polled within this window are considered abandoned and evicted. This +/// mirrors go-ethereum's default filter deadline and bounds the memory an idle client can retain. +const FILTER_TIMEOUT: Duration = Duration::from_secs(5 * 60); + +/// What a registered filter reports. +enum RegisteredFilter { + /// A log filter. The original [`Filter`] is kept so `eth_getFilterLogs` can replay the whole + /// range and `eth_getFilterChanges` can reuse its address/topics selection. + Logs(Filter), + /// A block filter, reporting the hashes of blocks added since the last poll. + Block, +} + +/// A registered filter together with the bookkeeping needed to serve incremental polls. +struct FilterEntry { + kind: RegisteredFilter, + /// Next block number (inclusive) that a poll should start reporting from. + next_from: SubstrateBlockNumber, + /// Inclusive upper bound for a log filter that pinned a concrete `toBlock`; `None` means the + /// filter follows the chain head. + to: Option, + /// Last time the filter was polled, used to evict abandoned filters. + last_poll: Instant, +} + +/// A snapshot of a filter taken while polling, so the (async) log or block lookup can run without +/// holding the registry lock across an `.await`. +pub enum FilterPoll { + /// Poll a log filter, reporting from `from` up to the (head-capped) upper bound `to`. + Logs { filter: Filter, from: SubstrateBlockNumber, to: Option }, + /// Poll a block filter, reporting block hashes from `from` onwards. + Block { from: SubstrateBlockNumber }, +} + +/// Thread-safe registry of installed filters, shared by the RPC server across all connections. +#[derive(Clone)] +pub struct FilterManager { + inner: Arc>>, + timeout: Duration, +} + +impl Default for FilterManager { + fn default() -> Self { + Self::new() + } +} + +impl FilterManager { + /// Create an empty registry using the default filter timeout. + pub fn new() -> Self { + Self { inner: Arc::new(Mutex::new(HashMap::new())), timeout: FILTER_TIMEOUT } + } + + fn filters(&self) -> MutexGuard<'_, HashMap> { + // The guarded sections only touch a `HashMap` and never panic, so the lock cannot actually + // be poisoned; recover the guard regardless to keep the RPC handlers panic-free. + self.inner.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Install a log filter reporting from `next_from`, optionally bounded above by `to`. + pub fn install_logs( + &self, + filter: Filter, + next_from: SubstrateBlockNumber, + to: Option, + ) -> U256 { + self.install(FilterEntry { + kind: RegisteredFilter::Logs(filter), + next_from, + to, + last_poll: Instant::now(), + }) + } + + /// Install a block filter reporting from `next_from`. + pub fn install_block(&self, next_from: SubstrateBlockNumber) -> U256 { + self.install(FilterEntry { + kind: RegisteredFilter::Block, + next_from, + to: None, + last_poll: Instant::now(), + }) + } + + fn install(&self, entry: FilterEntry) -> U256 { + let mut filters = self.filters(); + filters.retain(|_, entry| entry.last_poll.elapsed() <= self.timeout); + // Use an unpredictable id so a client cannot enumerate or tamper with filters it does not + // own, matching go-ethereum's random filter ids. + let id = loop { + let candidate = U256::from(rand::random::()); + if !filters.contains_key(&candidate) { + break candidate; + } + }; + filters.insert(id, entry); + id + } + + /// Snapshot a filter for polling and refresh its deadline. Returns `None` (removing the filter) + /// if it is unknown or has expired. + pub fn poll(&self, id: U256) -> Option { + let mut filters = self.filters(); + if filters.get(&id)?.last_poll.elapsed() > self.timeout { + filters.remove(&id); + return None; + } + let entry = filters.get_mut(&id).expect("entry was present on the line above; qed"); + entry.last_poll = Instant::now(); + Some(match &entry.kind { + RegisteredFilter::Logs(filter) => { + FilterPoll::Logs { filter: filter.clone(), from: entry.next_from, to: entry.to } + }, + RegisteredFilter::Block => FilterPoll::Block { from: entry.next_from }, + }) + } + + /// Advance a filter's cursor after a successful poll. A no-op if the filter is gone. + pub fn advance(&self, id: U256, next_from: SubstrateBlockNumber) { + if let Some(entry) = self.filters().get_mut(&id) { + entry.next_from = next_from; + } + } + + /// Return the original [`Filter`] of a log filter and refresh its deadline. Returns `None` for + /// an unknown, expired, or non-log filter. + pub fn logs_filter(&self, id: U256) -> Option { + let mut filters = self.filters(); + if filters.get(&id)?.last_poll.elapsed() > self.timeout { + filters.remove(&id); + return None; + } + let entry = filters.get_mut(&id).expect("entry was present on the line above; qed"); + entry.last_poll = Instant::now(); + match &entry.kind { + RegisteredFilter::Logs(filter) => Some(filter.clone()), + RegisteredFilter::Block => None, + } + } + + /// Remove a filter, returning whether it existed. + pub fn uninstall(&self, id: U256) -> bool { + self.filters().remove(&id).is_some() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn manager() -> FilterManager { + FilterManager::new() + } + + #[test] + fn install_returns_unique_ids() { + let manager = manager(); + let a = manager.install_block(0); + let b = manager.install_block(0); + assert_ne!(a, b, "each installed filter should get a distinct id"); + } + + #[test] + fn poll_log_filter_reports_cursor_then_advances() { + let manager = manager(); + let id = manager.install_logs(Filter::new(), 10, Some(20)); + + let Some(FilterPoll::Logs { from, to, .. }) = manager.poll(id) else { + panic!("expected a log poll"); + }; + assert_eq!(from, 10); + assert_eq!(to, Some(20)); + + // The cursor only moves once the caller has served the range. + manager.advance(id, 21); + let Some(FilterPoll::Logs { from, .. }) = manager.poll(id) else { + panic!("expected a log poll"); + }; + assert_eq!(from, 21); + } + + #[test] + fn poll_block_filter_reports_cursor() { + let manager = manager(); + let id = manager.install_block(5); + let Some(FilterPoll::Block { from }) = manager.poll(id) else { + panic!("expected a block poll"); + }; + assert_eq!(from, 5); + } + + #[test] + fn logs_filter_returns_none_for_block_filter() { + let manager = manager(); + let id = manager.install_block(0); + assert!(manager.logs_filter(id).is_none(), "block filters have no replayable log range"); + } + + #[test] + fn unknown_filter_is_not_found() { + let manager = manager(); + assert!(manager.poll(U256::from(42u64)).is_none()); + assert!(manager.logs_filter(U256::from(42u64)).is_none()); + assert!(!manager.uninstall(U256::from(42u64))); + } + + #[test] + fn uninstall_removes_filter() { + let manager = manager(); + let id = manager.install_logs(Filter::new(), 0, None); + assert!(manager.uninstall(id)); + assert!(!manager.uninstall(id), "a filter can only be uninstalled once"); + assert!(manager.poll(id).is_none(), "a polled-after-uninstall filter is gone"); + } + + #[test] + fn expired_filter_is_evicted_on_poll() { + let manager = + FilterManager { inner: Arc::new(Mutex::new(HashMap::new())), timeout: Duration::ZERO }; + let id = manager.install_block(0); + // With a zero timeout any elapsed time counts as expired. + std::thread::sleep(Duration::from_millis(1)); + assert!(manager.poll(id).is_none(), "an expired filter should not be polled"); + assert!(!manager.uninstall(id), "an expired filter should already be evicted"); + } +} diff --git a/substrate/frame/revive/rpc/src/lib.rs b/substrate/frame/revive/rpc/src/lib.rs index a32f338fb05b..ad61555aa1db 100644 --- a/substrate/frame/revive/rpc/src/lib.rs +++ b/substrate/frame/revive/rpc/src/lib.rs @@ -18,7 +18,7 @@ #![cfg_attr(docsrs, feature(doc_cfg))] pub use alloy_rpc_types::{BlockId, BlockNumberOrTag, Filter, FilterBlockOption}; -use client::ClientError; +use client::{ClientError, SubstrateBlockNumber}; use futures::{Stream, StreamExt, TryStreamExt}; use jsonrpsee::{ PendingSubscriptionSink, SubscriptionMessage, SubscriptionSink, @@ -58,6 +58,9 @@ pub use fee_history_provider::*; mod receipt_extractor; pub use receipt_extractor::*; +mod filters; +pub use filters::*; + mod apis; pub use apis::*; @@ -79,6 +82,9 @@ pub struct EthRpcServerImpl { /// When true, estimate_gas uses Pending block if no block is specified. use_pending_for_estimate_gas: bool, + + /// Registry of installed polling filters (`eth_newFilter` / `eth_newBlockFilter`). + filters: FilterManager, } impl EthRpcServerImpl { @@ -89,6 +95,7 @@ impl EthRpcServerImpl { accounts: vec![], allow_unprotected_txs: false, use_pending_for_estimate_gas: false, + filters: FilterManager::new(), } } @@ -135,6 +142,9 @@ pub enum EthRpcError { /// Received an invalid transaction #[error("Invalid transaction {0:?}")] TransactionTypeNotSupported(Byte), + /// No filter is registered under the given id (it is unknown or has expired). + #[error("filter not found")] + FilterNotFound, } impl From for ErrorObjectOwned { @@ -150,6 +160,7 @@ impl From for ErrorObjectOwned { EthRpcError::InvalidSignature | EthRpcError::AccountNotFound(_) | EthRpcError::InvalidTransaction | + EthRpcError::FilterNotFound | EthRpcError::TransactionTypeNotSupported(_) => CALL_EXECUTION_FAILED_CODE, }; Self::owned::(code, message, None) @@ -433,6 +444,70 @@ impl EthRpcServer for EthRpcServerImpl { Ok(FilterResults::Logs(logs)) } + async fn new_filter(&self, filter: Filter) -> RpcResult { + let head = self.client.block_number().await?; + let (from, to) = self.resolve_filter_range(&filter, head).await?; + Ok(self.filters.install_logs(filter, from, to)) + } + + async fn new_block_filter(&self) -> RpcResult { + let head = self.client.block_number().await?; + // Report only blocks added after the filter is installed, matching go-ethereum. + Ok(self.filters.install_block(head.saturating_add(1))) + } + + async fn get_filter_changes(&self, filter_id: U256) -> RpcResult { + match self.filters.poll(filter_id).ok_or(EthRpcError::FilterNotFound)? { + FilterPoll::Logs { mut filter, from, to } => { + let head = self.client.block_number().await?; + let upper = to.unwrap_or(head).min(head); + if from > upper { + return Ok(FilterResults::Logs(vec![])); + } + + filter.block_option = FilterBlockOption::Range { + from_block: Some(BlockNumberOrTag::Number(from)), + to_block: Some(BlockNumberOrTag::Number(upper)), + }; + let logs = self.client.logs(Some(filter)).await?; + self.filters.advance(filter_id, upper.saturating_add(1)); + Ok(FilterResults::Logs(logs)) + }, + FilterPoll::Block { from } => { + let head = self.client.block_number().await?; + if from > head { + return Ok(FilterResults::Hashes(vec![])); + } + + let mut hashes = Vec::new(); + for number in from..=head { + let Some(block) = self + .client + .block_by_number_or_tag(&BlockNumberOrTag::Number(number)) + .await? + else { + continue; + }; + if let Some(eth_block) = self.client.evm_block(block, false).await { + hashes.push(eth_block.hash); + } + } + self.filters.advance(filter_id, head.saturating_add(1)); + Ok(FilterResults::Hashes(hashes)) + }, + } + } + + async fn get_filter_logs(&self, filter_id: U256) -> RpcResult { + let filter = self.filters.logs_filter(filter_id).ok_or(EthRpcError::FilterNotFound)?; + let logs = self.client.logs(Some(filter)).await?; + Ok(FilterResults::Logs(logs)) + } + + async fn uninstall_filter(&self, filter_id: U256) -> RpcResult { + Ok(self.filters.uninstall(filter_id)) + } + async fn get_storage_at( &self, address: H160, @@ -559,6 +634,69 @@ impl EthRpcServer for EthRpcServerImpl { } impl EthRpcServerImpl { + /// Resolve the block range of a newly installed log filter into concrete block numbers. + /// + /// Returns `(from, to)` where `from` is the first block a poll should report and `to` is the + /// inclusive upper bound (`None` to follow the chain head). A filter without an explicit + /// `fromBlock`, or one pinned to `latest`/`pending`, reports only blocks added after it was + /// installed, matching go-ethereum. + async fn resolve_filter_range( + &self, + filter: &Filter, + head: SubstrateBlockNumber, + ) -> RpcResult<(SubstrateBlockNumber, Option)> { + match &filter.block_option { + FilterBlockOption::AtBlockHash(block_hash) => { + let number = self + .resolve_ethereum_block_number(H256::from_slice(block_hash.as_slice())) + .await?; + Ok((number, Some(number))) + }, + FilterBlockOption::Range { from_block, to_block } => { + let from = match from_block { + None | Some(BlockNumberOrTag::Latest) | Some(BlockNumberOrTag::Pending) => { + head.saturating_add(1) + }, + Some(tag) => self.resolve_block_tag(tag, head).await?, + }; + let to = match to_block { + None | Some(BlockNumberOrTag::Latest) | Some(BlockNumberOrTag::Pending) => None, + Some(tag) => Some(self.resolve_block_tag(tag, head).await?), + }; + Ok((from, to)) + }, + } + } + + /// Resolve a block tag (a number, or `earliest`/`finalized`/`safe`) into a concrete block + /// number, falling back to the chain head if the tagged block cannot be located. + async fn resolve_block_tag( + &self, + tag: &BlockNumberOrTag, + head: SubstrateBlockNumber, + ) -> RpcResult { + if let BlockNumberOrTag::Number(number) = tag { + return Ok(*number); + } + match self.client.block_by_number_or_tag(tag).await? { + Some(block) => Ok(block.number()), + None => Ok(head), + } + } + + /// Resolve an Ethereum block hash to its block number, erroring if the block is unknown. + async fn resolve_ethereum_block_number( + &self, + block_hash: H256, + ) -> RpcResult { + let block = self + .client + .block_by_ethereum_hash(&block_hash) + .await? + .ok_or(ClientError::BlockNotFound)?; + Ok(block.number()) + } + async fn get_transaction_by_substrate_block_hash_and_index( &self, substrate_block_hash: H256, @@ -633,6 +771,7 @@ mod error_codes_tests { EthRpcError::TransactionTypeNotSupported(Byte::from(0x7eu8)), CALL_EXECUTION_FAILED_CODE, ), + (EthRpcError::FilterNotFound, CALL_EXECUTION_FAILED_CODE), ]; for (err, expected_code) in cases { diff --git a/substrate/frame/revive/rpc/src/tests.rs b/substrate/frame/revive/rpc/src/tests.rs index c9ad1fedbe8c..d7cf238596ed 100644 --- a/substrate/frame/revive/rpc/src/tests.rs +++ b/substrate/frame/revive/rpc/src/tests.rs @@ -377,6 +377,7 @@ async fn run_all_eth_rpc_tests_inner() -> anyhow::Result<()> { test_trace_block_returns_v1_trace_on_v1_input_and_v2_trace_on_v2_input, test_transfer, test_deploy_and_call, + test_filter_lifecycle, test_receipt_mixed_revert_and_logs_same_block, test_runtime_api_dry_run_addr_works, test_invalid_transaction, @@ -681,6 +682,108 @@ async fn test_receipt_mixed_revert_and_logs_same_block() -> anyhow::Result<()> { Ok(()) } +/// `FilterResults` is an untagged enum, so an empty JSON array (`[]`) always deserializes to its +/// first variant (`Hashes`). Treat either empty variant as "no changes". +fn is_empty_filter_results(results: &FilterResults) -> bool { + match results { + FilterResults::Hashes(hashes) => hashes.is_empty(), + FilterResults::Logs(logs) => logs.is_empty(), + } +} + +/// Exercise the polling filter API end to end: `eth_newFilter`, `eth_getFilterChanges`, +/// `eth_getFilterLogs`, `eth_newBlockFilter` and `eth_uninstallFilter`. +async fn test_filter_lifecycle() -> anyhow::Result<()> { + let client = Arc::new(SharedResources::client().await); + let account = Account::default(); + + // Deploy a contract that emits `Received(address,uint256)` when it receives value. + let (bytes, _) = pallet_revive_fixtures::compile_module_with_type( + "SimpleReceiver", + pallet_revive_fixtures::FixtureType::Solc, + )?; + let nonce = client.get_transaction_count(account.address(), Default::default()).await?; + let tx = TransactionBuilder::new(client.clone()).input(bytes.to_vec()).send().await?; + let receipt = tx.wait_for_receipt().await?; + let contract_address = create1(&account.address(), nonce.try_into().unwrap()); + assert_eq!(Some(contract_address), receipt.contract_address); + + // Emit a single event. + let value = U256::from(1_000_000_000_000u128); + let tx = TransactionBuilder::new(client.clone()) + .value(value) + .to(contract_address) + .send() + .await?; + let emit_receipt = tx.wait_for_receipt().await?; + + // A log filter scoped to this (uniquely addressed) contract, starting at the emit block. + let filter = Filter::new() + .from_block(emit_receipt.block_number.as_u64()) + .address(AlloyAddress::from_slice(contract_address.as_ref())); + let filter_id = client.new_filter(filter).await?; + + // The first poll returns the emitted log. + let logs = match client.get_filter_changes(filter_id).await? { + FilterResults::Logs(logs) => logs, + other => return Err(anyhow!("Expected Logs from eth_getFilterChanges, got: {other:?}")), + }; + assert_eq!(logs.len(), 1, "filter should report exactly the one emitted log"); + let event_signature = H256(sp_io::hashing::keccak_256(b"Received(address,uint256)")); + assert_eq!(logs[0].address, contract_address); + assert!(!logs[0].topics.is_empty(), "log should carry the event signature topic"); + assert_eq!(logs[0].topics[0], event_signature); + assert_eq!(logs[0].transaction_hash, emit_receipt.transaction_hash); + + // `eth_getFilterLogs` replays the full range and returns the same log. + let all_logs = match client.get_filter_logs(filter_id).await? { + FilterResults::Logs(logs) => logs, + other => return Err(anyhow!("Expected Logs from eth_getFilterLogs, got: {other:?}")), + }; + assert_eq!(all_logs, logs, "getFilterLogs should replay the same log"); + + // A second poll has no new logs, as the cursor advanced past the emit block. + let empty = client.get_filter_changes(filter_id).await?; + assert!(is_empty_filter_results(&empty), "a second poll should be empty, got: {empty:?}"); + + // Uninstall succeeds once; afterwards the id is unknown. + assert!( + client.uninstall_filter(filter_id).await?, + "uninstall should report the filter existed" + ); + assert!(!client.uninstall_filter(filter_id).await?, "a filter can only be uninstalled once"); + let err = client + .get_filter_changes(filter_id) + .await + .expect_err("polling a removed filter should error"); + let ClientError::Call(call) = err else { + return Err(anyhow!("Expected a Call error, got: {err:?}")); + }; + assert_eq!(call.message(), "filter not found"); + + // A block filter reports the hashes of blocks added after it was installed. + let block_filter_id = client.new_block_filter().await?; + let tx = TransactionBuilder::new(client.clone()) + .value(value) + .to(contract_address) + .send() + .await?; + tx.wait_for_receipt().await?; + let block_hashes = match client.get_filter_changes(block_filter_id).await? { + FilterResults::Hashes(hashes) => hashes, + other => return Err(anyhow!("Expected Hashes from a block filter, got: {other:?}")), + }; + assert!(!block_hashes.is_empty(), "block filter should report the newly produced block"); + let newest = block_hashes.last().expect("non-empty checked above"); + assert!( + client.get_block_by_hash(*newest, false).await?.is_some(), + "a reported block hash should resolve to a real block" + ); + assert!(client.uninstall_filter(block_filter_id).await?); + + Ok(()) +} + async fn test_runtime_api_dry_run_addr_works() -> anyhow::Result<()> { let client = Arc::new(SharedResources::client().await); let node_client = SharedResources::node_client().await; From e75a8368f52c28f7d5bb2e56679f9e89d91c9b1d Mon Sep 17 00:00:00 2001 From: soloking1412 Date: Sat, 18 Jul 2026 23:25:34 +0530 Subject: [PATCH 2/3] prdoc: rename to PR number --- prdoc/{pr_0000.prdoc => pr_12685.prdoc} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename prdoc/{pr_0000.prdoc => pr_12685.prdoc} (100%) diff --git a/prdoc/pr_0000.prdoc b/prdoc/pr_12685.prdoc similarity index 100% rename from prdoc/pr_0000.prdoc rename to prdoc/pr_12685.prdoc From fb15bcf30dd2a20de58015fa4ab3cf580a617667 Mon Sep 17 00:00:00 2001 From: soloking1412 Date: Mon, 20 Jul 2026 22:45:25 +0530 Subject: [PATCH 3/3] =?UTF-8?q?pallet-revive:=20address=20review=20?= =?UTF-8?q?=E2=80=94=20back=20polling=20filters=20with=20the=20broadcast?= =?UTF-8?q?=20event=20system?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- prdoc/pr_12685.prdoc | 19 +- .../revive/rpc/src/apis/execution_apis.rs | 8 +- substrate/frame/revive/rpc/src/cli.rs | 23 +- substrate/frame/revive/rpc/src/filters.rs | 528 +++++++++++++----- substrate/frame/revive/rpc/src/lib.rs | 162 ++---- substrate/frame/revive/rpc/src/tests.rs | 90 ++- .../revive/rpc/src/types/subscription.rs | 23 + 7 files changed, 551 insertions(+), 302 deletions(-) diff --git a/prdoc/pr_12685.prdoc b/prdoc/pr_12685.prdoc index 3f596ee3625e..a51c1cef4ee5 100644 --- a/prdoc/pr_12685.prdoc +++ b/prdoc/pr_12685.prdoc @@ -1,7 +1,7 @@ title: 'pallet-revive: implement the eth_newFilter polling filter API' doc: -- audience: Node Dev +- audience: Runtime Dev description: |- The `pallet-revive` Ethereum JSON-RPC server only offered `eth_subscribe` (a WebSocket push API) for watching logs and new blocks, leaving the standard HTTP polling filter family @@ -17,13 +17,20 @@ doc: - `eth_getFilterLogs`: return all logs matching a log filter over its full range. - `eth_uninstallFilter`: remove a filter. - Filters are pull-based: each one only remembers the next block to report from, and the log - lookup reuses the existing `eth_getLogs` machinery. Ids are unpredictable (128-bit random) so - a client cannot enumerate or tamper with filters it does not own, and a filter that is not - polled within five minutes is evicted, mirroring go-ethereum and bounding memory use. + The polling filters and `eth_subscribe` are driven by the same block and log broadcast + channels, so both APIs report the same events and cannot diverge. A polling filter owns its + channel and drains it on each `eth_getFilterChanges`; a filter's `fromBlock`/`toBlock` bound + only the historical `eth_getFilterLogs` replay, which reuses the existing `eth_getLogs` + machinery. + + Memory is bounded on two axes: the number of installed filters is capped, and each filter + holds only a broadcast cursor into the shared, fixed-capacity ring buffer rather than its own + growing backlog. Ids are unpredictable (128-bit random) so a client cannot enumerate or tamper + with filters it does not own, and a background task evicts filters not polled within five + minutes as well as log filters whose fixed `toBlock` the chain has passed. `eth_newPendingTransactionFilter` is intentionally out of scope, as the server does not expose a pending-transaction pool (the same reason `eth_subscribe` has no `newPendingTransactions`). crates: - name: pallet-revive-eth-rpc - bump: minor + bump: major diff --git a/substrate/frame/revive/rpc/src/apis/execution_apis.rs b/substrate/frame/revive/rpc/src/apis/execution_apis.rs index 65c402eda8b5..0e1032ed7b73 100644 --- a/substrate/frame/revive/rpc/src/apis/execution_apis.rs +++ b/substrate/frame/revive/rpc/src/apis/execution_apis.rs @@ -109,14 +109,16 @@ pub trait EthRpc { #[method(name = "eth_newBlockFilter")] async fn new_block_filter(&self) -> RpcResult; - /// Polls the filter with the given id, returning the logs (for a log filter) or block hashes - /// (for a block filter) that occurred since the last poll. + /// Polls the filter with the given id, returning what changed since the last poll: matching + /// logs for a log filter, or block hashes for a block filter. The result is a union of the two + /// (`FilterResults`) because the id alone does not tell the caller which kind it is, mirroring + /// `eth_getFilterChanges`, which returns either shape. #[method(name = "eth_getFilterChanges")] async fn get_filter_changes(&self, filter_id: U256) -> RpcResult; /// Returns all logs matching the log filter with the given id, over its full range. #[method(name = "eth_getFilterLogs")] - async fn get_filter_logs(&self, filter_id: U256) -> RpcResult; + async fn get_filter_logs(&self, filter_id: U256) -> RpcResult>; /// Uninstalls the filter with the given id. Returns `true` if the filter existed. #[method(name = "eth_uninstallFilter")] diff --git a/substrate/frame/revive/rpc/src/cli.rs b/substrate/frame/revive/rpc/src/cli.rs index 613b1dc9f858..49899a5a7e46 100644 --- a/substrate/frame/revive/rpc/src/cli.rs +++ b/substrate/frame/revive/rpc/src/cli.rs @@ -16,8 +16,8 @@ // limitations under the License. //! The Ethereum JSON-RPC server. use crate::{ - DbContext, DebugRpcServer, DebugRpcServerImpl, EthRpcServer, EthRpcServerImpl, LOG_TARGET, - PolkadotRpcServer, PolkadotRpcServerImpl, ReceiptExtractor, ReceiptProvider, + DbContext, DebugRpcServer, DebugRpcServerImpl, EthRpcServer, EthRpcServerImpl, FilterManager, + LOG_TARGET, PolkadotRpcServer, PolkadotRpcServerImpl, ReceiptExtractor, ReceiptProvider, SubxtBlockInfoProvider, SystemHealthRpcServer, SystemHealthRpcServerImpl, client::{Client, ClientError, SubscriptionGapQueue, SubscriptionType, connect}, }; @@ -402,7 +402,11 @@ pub fn run(cmd: CliCommand) -> anyhow::Result<()> { let rpc_runtime = create_rpc_runtime(rpc_config.max_connections) .map_err(|e| anyhow::anyhow!("Failed to create RPC runtime: {}", e))?; - let rpc_api = rpc_module(is_dev, client.clone(), allow_unprotected_txs)?; + // Share one filter registry between the RPC handlers and the background task that periodically + // evicts expired and no-longer-applicable filters. + let filter_registry = FilterManager::new(); + let rpc_api = + rpc_module(is_dev, client.clone(), allow_unprotected_txs, filter_registry.clone())?; let rpc_server_handle = start_rpc_servers( &rpc_config, prometheus_registry, @@ -412,6 +416,17 @@ pub fn run(cmd: CliCommand) -> anyhow::Result<()> { None, )?; + // Periodically evict abandoned and no-longer-applicable polling filters. + let cleanup_client = client.clone(); + task_manager.spawn_essential_handle().spawn( + "eth-rpc-filter-cleanup", + None, + filter_registry.run_cleanup(move || { + let client = cleanup_client.clone(); + async move { client.block_number().await.ok() } + }), + ); + task_manager .spawn_essential_handle() .spawn("block-subscription", None, async move { @@ -446,8 +461,10 @@ fn rpc_module( is_dev: bool, client: Client, allow_unprotected_txs: bool, + filter_registry: FilterManager, ) -> Result, sc_service::Error> { let eth_api = EthRpcServerImpl::new(client.clone()) + .with_filter_registry(filter_registry) .with_accounts(if is_dev { vec![ crate::Account::from(subxt_signer::eth::dev::alith()), diff --git a/substrate/frame/revive/rpc/src/filters.rs b/substrate/frame/revive/rpc/src/filters.rs index 1ff341d2156b..cd2ca0482e3b 100644 --- a/substrate/frame/revive/rpc/src/filters.rs +++ b/substrate/frame/revive/rpc/src/filters.rs @@ -18,57 +18,200 @@ //! In-memory registry backing the polling filter API (`eth_newFilter`, `eth_newBlockFilter`, //! `eth_getFilterChanges`, `eth_getFilterLogs`, `eth_uninstallFilter`). //! -//! Unlike `eth_subscribe`, these methods are pull-based: a client installs a filter and later -//! polls it for whatever changed since the previous poll. The registry therefore only needs to -//! remember, per filter, the next block to report from; the actual log lookup is served on demand -//! by reusing [`crate::client::Client::logs`]. +//! Both `eth_subscribe` and this polling API are driven by the same source of truth: the block and +//! log broadcast channels the client publishes every imported block and log to (see +//! [`crate::client::Client::get_block_subscription_rx`] and +//! [`crate::client::Client::get_log_subscription_rx`]). `eth_subscribe` consumes a channel through +//! a jsonrpsee sink; a polling filter instead *owns* its channel here and drains it on each +//! `eth_getFilterChanges`. Sharing one event source keeps the two APIs from diverging. +//! +//! Memory is bounded on two axes: +//! - the number of installed filters is capped at [`MAX_FILTERS`], so a client cannot exhaust +//! memory by opening filters it never polls or uninstalls; +//! - each filter holds only a broadcast *cursor* (a handle into the shared, fixed-capacity ring +//! buffer) rather than its own growing backlog, so a million filters cost a million cursors and +//! one shared buffer, not a million buffers. +//! +//! A filter not polled within [`FILTER_TIMEOUT`] is evicted by a background task ([`run_cleanup`]), +//! as are log filters whose fixed `toBlock` the chain has already passed. The trade-off versus +//! go-ethereum's unbounded per-filter backlog is that a filter polled too slowly can miss events +//! (its cursor lags the ring and the channel reports `Lagged`); bounded memory is preferred here to +//! unbounded growth. +//! +//! [`run_cleanup`]: FilterManager::run_cleanup use crate::{client::SubstrateBlockNumber, *}; use std::{ collections::HashMap, - sync::{Arc, Mutex, MutexGuard}, + fmt, + sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}, time::{Duration, Instant}, }; +use tokio::sync::broadcast; -/// Filters that are not polled within this window are considered abandoned and evicted. This -/// mirrors go-ethereum's default filter deadline and bounds the memory an idle client can retain. +/// Filters not polled within this window are considered abandoned and evicted, mirroring +/// go-ethereum's default filter deadline. const FILTER_TIMEOUT: Duration = Duration::from_secs(5 * 60); -/// What a registered filter reports. -enum RegisteredFilter { - /// A log filter. The original [`Filter`] is kept so `eth_getFilterLogs` can replay the whole - /// range and `eth_getFilterChanges` can reuse its address/topics selection. - Logs(Filter), - /// A block filter, reporting the hashes of blocks added since the last poll. - Block, +/// How often the background task sweeps expired and no-longer-applicable filters. +const CLEANUP_INTERVAL: Duration = Duration::from_secs(60); + +/// Upper bound on concurrently installed filters. Installing beyond this is rejected so that a +/// client cannot exhaust memory by opening filters without ever polling or uninstalling them. +pub const MAX_FILTERS: usize = 8_192; + +/// Identifier handed to a client for an installed filter. +/// +/// A newtype rather than a bare [`U256`] so the filter-manager interface cannot be confused with +/// the many other `U256`-typed values in the eth-rpc, and so that ids are minted in exactly one +/// place. +#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] +pub struct FilterId(U256); + +impl FilterId { + /// Mint an unpredictable id. 128 random bits make ids impractical to guess, so a client cannot + /// enumerate or tamper with filters it does not own; go-ethereum uses random ids for the same + /// reason. + fn random() -> Self { + Self(U256::from(rand::random::())) + } +} + +impl From for FilterId { + fn from(value: U256) -> Self { + Self(value) + } +} + +impl From for U256 { + fn from(value: FilterId) -> Self { + value.0 + } +} + +impl fmt::Display for FilterId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Installing a filter failed because [`MAX_FILTERS`] are already installed. +#[derive(Debug, PartialEq, Eq)] +pub struct TooManyFilters; + +/// Poll state mutated on every `eth_getFilterChanges`. Guarded on its own so concurrent polls of +/// different filters only need a shared read lock on the registry. +enum PollState { + /// A log filter: drains streamed logs and keeps those the `matcher` selects, using the same + /// address/topics logic as `eth_subscribe`. + Logs { receiver: broadcast::Receiver, matcher: LogsSubscriptionFilter, last_poll: Instant }, + /// A block filter: drains and reports the hashes of blocks imported since the previous poll. + Block { receiver: broadcast::Receiver, last_poll: Instant }, } -/// A registered filter together with the bookkeeping needed to serve incremental polls. +impl PollState { + fn last_poll(&self) -> Instant { + match self { + PollState::Logs { last_poll, .. } | PollState::Block { last_poll, .. } => *last_poll, + } + } + + /// Refresh the deadline; called whenever the filter is successfully polled. + fn touch(&mut self) { + let now = Instant::now(); + match self { + PollState::Logs { last_poll, .. } | PollState::Block { last_poll, .. } => { + *last_poll = now + }, + } + } + + /// Drain everything reported since the previous poll. + fn drain(&mut self) -> FilterResults { + match self { + PollState::Logs { receiver, matcher, .. } => { + let mut logs = Vec::new(); + drain_receiver(receiver, "log", |log| { + if matcher.matches(&log) { + logs.push(log); + } + }); + FilterResults::Logs(logs) + }, + PollState::Block { receiver, .. } => { + let mut hashes = Vec::new(); + drain_receiver(receiver, "block", |block| hashes.push(block.hash)); + FilterResults::Hashes(hashes) + }, + } + } +} + +/// Drain a broadcast receiver of everything currently buffered, handing each item to `on_item`. +/// +/// A `Lagged` cursor means the shared ring advanced past this filter (it was polled too slowly) and +/// the skipped items are lost for it; that is the deliberate cost of bounding memory rather than +/// keeping an unbounded per-filter backlog. Draining continues past the gap. +fn drain_receiver( + receiver: &mut broadcast::Receiver, + kind: &str, + mut on_item: impl FnMut(T), +) { + loop { + match receiver.try_recv() { + Ok(item) => on_item(item), + Err(broadcast::error::TryRecvError::Empty) | + Err(broadcast::error::TryRecvError::Closed) => break, + Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + log::warn!(target: LOG_TARGET, "{kind} filter lagged; {skipped} events dropped"); + }, + } + } +} + +/// An installed filter: its immutable definition plus the mutable poll state. struct FilterEntry { - kind: RegisteredFilter, - /// Next block number (inclusive) that a poll should start reporting from. - next_from: SubstrateBlockNumber, - /// Inclusive upper bound for a log filter that pinned a concrete `toBlock`; `None` means the - /// filter follows the chain head. - to: Option, - /// Last time the filter was polled, used to evict abandoned filters. - last_poll: Instant, + /// The original request of a log filter, kept so `eth_getFilterLogs` can replay its whole + /// range and cleanup can test its `toBlock`; `None` for a block filter. Immutable, so it is + /// read directly under the registry lock without touching the per-entry [`Mutex`]. + filter: Option, + state: Mutex, } -/// A snapshot of a filter taken while polling, so the (async) log or block lookup can run without -/// holding the registry lock across an `.await`. -pub enum FilterPoll { - /// Poll a log filter, reporting from `from` up to the (head-capped) upper bound `to`. - Logs { filter: Filter, from: SubstrateBlockNumber, to: Option }, - /// Poll a block filter, reporting block hashes from `from` onwards. - Block { from: SubstrateBlockNumber }, +impl FilterEntry { + /// Whether to keep this filter during a cleanup sweep at the given chain head. + /// + /// Drops filters not polled within [`FILTER_TIMEOUT`], and — when `head` is known — log filters + /// whose fixed `toBlock` the head has already passed, since no future streamed log can fall in + /// their range and so a poll could only ever return empty. + fn should_retain(&self, head: Option) -> bool { + let last_poll = self.state.lock().unwrap_or_else(|p| p.into_inner()).last_poll(); + if last_poll.elapsed() > FILTER_TIMEOUT { + return false; + } + if let (Some(head), Some(filter)) = (head, &self.filter) { + if let FilterBlockOption::Range { + to_block: Some(BlockNumberOrTag::Number(to)), .. + } = &filter.block_option + { + if *to < head { + return false; + } + } + } + true + } } /// Thread-safe registry of installed filters, shared by the RPC server across all connections. +/// +/// The registry lock is an [`RwLock`] because polls (reads) vastly outnumber installs and +/// uninstalls (writes): a poll takes the read lock to find its entry and then a per-entry +/// [`Mutex`], so different filters poll concurrently, while only structural changes take the write +/// lock. #[derive(Clone)] pub struct FilterManager { - inner: Arc>>, - timeout: Duration, + inner: Arc>>, } impl Default for FilterManager { @@ -78,101 +221,121 @@ impl Default for FilterManager { } impl FilterManager { - /// Create an empty registry using the default filter timeout. + /// Create an empty registry. pub fn new() -> Self { - Self { inner: Arc::new(Mutex::new(HashMap::new())), timeout: FILTER_TIMEOUT } + Self { inner: Arc::new(RwLock::new(HashMap::new())) } } - fn filters(&self) -> MutexGuard<'_, HashMap> { - // The guarded sections only touch a `HashMap` and never panic, so the lock cannot actually - // be poisoned; recover the guard regardless to keep the RPC handlers panic-free. - self.inner.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + fn read(&self) -> RwLockReadGuard<'_, HashMap> { + // The guarded sections never panic, so the lock cannot actually be poisoned; recover the + // guard regardless to keep the RPC handlers panic-free. + self.inner.read().unwrap_or_else(|poisoned| poisoned.into_inner()) } - /// Install a log filter reporting from `next_from`, optionally bounded above by `to`. + fn write(&self) -> RwLockWriteGuard<'_, HashMap> { + self.inner.write().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Install a log filter fed by `receiver`, selecting streamed logs by `filter`'s address and + /// topics. Fails if [`MAX_FILTERS`] are already installed. pub fn install_logs( &self, filter: Filter, - next_from: SubstrateBlockNumber, - to: Option, - ) -> U256 { - self.install(FilterEntry { - kind: RegisteredFilter::Logs(filter), - next_from, - to, - last_poll: Instant::now(), - }) - } - - /// Install a block filter reporting from `next_from`. - pub fn install_block(&self, next_from: SubstrateBlockNumber) -> U256 { - self.install(FilterEntry { - kind: RegisteredFilter::Block, - next_from, - to: None, - last_poll: Instant::now(), - }) - } - - fn install(&self, entry: FilterEntry) -> U256 { - let mut filters = self.filters(); - filters.retain(|_, entry| entry.last_poll.elapsed() <= self.timeout); - // Use an unpredictable id so a client cannot enumerate or tamper with filters it does not - // own, matching go-ethereum's random filter ids. + receiver: broadcast::Receiver, + ) -> Result { + let matcher = LogsSubscriptionFilter::from_filter(&filter); + self.install(Some(filter), PollState::Logs { receiver, matcher, last_poll: Instant::now() }) + } + + /// Install a block filter fed by `receiver`. Only blocks imported after installation are + /// reported, because the receiver starts at the current channel head — there is no manual + /// `head + 1` bookkeeping. Fails if [`MAX_FILTERS`] are already installed. + pub fn install_block( + &self, + receiver: broadcast::Receiver, + ) -> Result { + self.install(None, PollState::Block { receiver, last_poll: Instant::now() }) + } + + fn install( + &self, + filter: Option, + state: PollState, + ) -> Result { + let mut filters = self.write(); + if filters.len() >= MAX_FILTERS { + return Err(TooManyFilters); + } let id = loop { - let candidate = U256::from(rand::random::()); + let candidate = FilterId::random(); if !filters.contains_key(&candidate) { break candidate; } }; - filters.insert(id, entry); - id + filters.insert(id, FilterEntry { filter, state: Mutex::new(state) }); + Ok(id) } - /// Snapshot a filter for polling and refresh its deadline. Returns `None` (removing the filter) - /// if it is unknown or has expired. - pub fn poll(&self, id: U256) -> Option { - let mut filters = self.filters(); - if filters.get(&id)?.last_poll.elapsed() > self.timeout { - filters.remove(&id); - return None; + /// Drain everything reported since the previous poll: matching logs for a log filter, or block + /// hashes for a block filter. Returns `None` (evicting the filter) if it is unknown or has not + /// been polled within [`FILTER_TIMEOUT`]. + pub fn poll_changes(&self, id: FilterId) -> Option { + { + let filters = self.read(); + let entry = filters.get(&id)?; + let mut state = entry.state.lock().unwrap_or_else(|p| p.into_inner()); + if state.last_poll().elapsed() <= FILTER_TIMEOUT { + state.touch(); + return Some(state.drain()); + } } - let entry = filters.get_mut(&id).expect("entry was present on the line above; qed"); - entry.last_poll = Instant::now(); - Some(match &entry.kind { - RegisteredFilter::Logs(filter) => { - FilterPoll::Logs { filter: filter.clone(), from: entry.next_from, to: entry.to } - }, - RegisteredFilter::Block => FilterPoll::Block { from: entry.next_from }, - }) + // Expired: escalate to the write lock and remove it, matching go-ethereum, which treats a + // filter past its deadline as gone. + self.write().remove(&id); + None } - /// Advance a filter's cursor after a successful poll. A no-op if the filter is gone. - pub fn advance(&self, id: U256, next_from: SubstrateBlockNumber) { - if let Some(entry) = self.filters().get_mut(&id) { - entry.next_from = next_from; + /// Return the original [`Filter`] of a log filter, so `eth_getFilterLogs` can replay its whole + /// range. Returns `None` for an unknown, expired, or block filter. + pub fn logs_filter(&self, id: FilterId) -> Option { + { + let filters = self.read(); + let entry = filters.get(&id)?; + let mut state = entry.state.lock().unwrap_or_else(|p| p.into_inner()); + if state.last_poll().elapsed() <= FILTER_TIMEOUT { + state.touch(); + return entry.filter.clone(); + } } + self.write().remove(&id); + None } - /// Return the original [`Filter`] of a log filter and refresh its deadline. Returns `None` for - /// an unknown, expired, or non-log filter. - pub fn logs_filter(&self, id: U256) -> Option { - let mut filters = self.filters(); - if filters.get(&id)?.last_poll.elapsed() > self.timeout { - filters.remove(&id); - return None; - } - let entry = filters.get_mut(&id).expect("entry was present on the line above; qed"); - entry.last_poll = Instant::now(); - match &entry.kind { - RegisteredFilter::Logs(filter) => Some(filter.clone()), - RegisteredFilter::Block => None, - } + /// Remove a filter, returning whether it existed. + pub fn uninstall(&self, id: FilterId) -> bool { + self.write().remove(&id).is_some() } - /// Remove a filter, returning whether it existed. - pub fn uninstall(&self, id: U256) -> bool { - self.filters().remove(&id).is_some() + /// Sweep expired and no-longer-applicable filters. `head`, when known, additionally drops log + /// filters whose fixed `toBlock` the chain has passed. + pub fn evict(&self, head: Option) { + self.write().retain(|_, entry| entry.should_retain(head)); + } + + /// Run the periodic cleanup loop; intended to be spawned as a long-lived task. `current_head` + /// yields the latest block number (or `None` while momentarily unavailable) for staleness + /// checks. + pub async fn run_cleanup(self, current_head: F) + where + F: Fn() -> Fut, + Fut: std::future::Future>, + { + let mut interval = tokio::time::interval(CLEANUP_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + interval.tick().await; + self.evict(current_head().await); + } } } @@ -180,79 +343,148 @@ impl FilterManager { mod tests { use super::*; - fn manager() -> FilterManager { - FilterManager::new() + fn log_from(address: u8) -> Log { + Log { address: H160::repeat_byte(address), ..Default::default() } + } + + fn block_with_hash(hash: u8) -> Block { + Block { hash: H256::repeat_byte(hash), ..Default::default() } + } + + /// A log filter matching a single emitter address. + fn filter_for(address: u8) -> Filter { + Filter::new().address(alloy_primitives::Address::from([address; 20])) + } + + /// Push `last_poll` far enough into the past that the filter counts as expired. + fn expire(manager: &FilterManager, id: FilterId) { + let filters = manager.write(); + let mut state = filters.get(&id).expect("filter exists").state.lock().unwrap(); + let past = Instant::now().checked_sub(FILTER_TIMEOUT * 2).expect("test clock in range"); + match &mut *state { + PollState::Logs { last_poll, .. } | PollState::Block { last_poll, .. } => { + *last_poll = past + }, + } } #[test] fn install_returns_unique_ids() { - let manager = manager(); - let a = manager.install_block(0); - let b = manager.install_block(0); + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(4); + let a = manager.install_block(tx.subscribe()).unwrap(); + let b = manager.install_block(tx.subscribe()).unwrap(); assert_ne!(a, b, "each installed filter should get a distinct id"); } #[test] - fn poll_log_filter_reports_cursor_then_advances() { - let manager = manager(); - let id = manager.install_logs(Filter::new(), 10, Some(20)); + fn unknown_filter_is_not_found() { + let manager = FilterManager::new(); + let id = FilterId::from(U256::from(42u64)); + assert!(manager.poll_changes(id).is_none()); + assert!(manager.logs_filter(id).is_none()); + assert!(!manager.uninstall(id)); + } + + #[test] + fn uninstall_removes_filter() { + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(4); + let id = manager.install_logs(Filter::new(), tx.subscribe()).unwrap(); + assert!(manager.uninstall(id)); + assert!(!manager.uninstall(id), "a filter can only be uninstalled once"); + assert!(manager.poll_changes(id).is_none(), "a polled-after-uninstall filter is gone"); + } + + #[test] + fn log_filter_reports_only_matching_logs() { + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(8); + let id = manager.install_logs(filter_for(0xAA), tx.subscribe()).unwrap(); + + // Only logs sent after the filter subscribed are seen, and only matching ones are kept. + tx.send(log_from(0xAA)).unwrap(); + tx.send(log_from(0xBB)).unwrap(); + tx.send(log_from(0xAA)).unwrap(); - let Some(FilterPoll::Logs { from, to, .. }) = manager.poll(id) else { - panic!("expected a log poll"); + let Some(FilterResults::Logs(logs)) = manager.poll_changes(id) else { + panic!("expected logs"); }; - assert_eq!(from, 10); - assert_eq!(to, Some(20)); + assert_eq!(logs.len(), 2, "only the two 0xAA logs match the filter"); + assert!(logs.iter().all(|l| l.address == H160::repeat_byte(0xAA))); - // The cursor only moves once the caller has served the range. - manager.advance(id, 21); - let Some(FilterPoll::Logs { from, .. }) = manager.poll(id) else { - panic!("expected a log poll"); + // A second poll with nothing new returns an empty set rather than the previous logs. + let Some(FilterResults::Logs(logs)) = manager.poll_changes(id) else { + panic!("expected logs"); }; - assert_eq!(from, 21); + assert!(logs.is_empty(), "changes since the last poll should be empty"); } #[test] - fn poll_block_filter_reports_cursor() { - let manager = manager(); - let id = manager.install_block(5); - let Some(FilterPoll::Block { from }) = manager.poll(id) else { - panic!("expected a block poll"); + fn block_filter_reports_new_block_hashes() { + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(8); + let id = manager.install_block(tx.subscribe()).unwrap(); + + tx.send(block_with_hash(1)).unwrap(); + tx.send(block_with_hash(2)).unwrap(); + + let Some(FilterResults::Hashes(hashes)) = manager.poll_changes(id) else { + panic!("expected hashes"); }; - assert_eq!(from, 5); + assert_eq!(hashes, vec![H256::repeat_byte(1), H256::repeat_byte(2)]); } #[test] fn logs_filter_returns_none_for_block_filter() { - let manager = manager(); - let id = manager.install_block(0); + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(4); + let id = manager.install_block(tx.subscribe()).unwrap(); assert!(manager.logs_filter(id).is_none(), "block filters have no replayable log range"); } #[test] - fn unknown_filter_is_not_found() { - let manager = manager(); - assert!(manager.poll(U256::from(42u64)).is_none()); - assert!(manager.logs_filter(U256::from(42u64)).is_none()); - assert!(!manager.uninstall(U256::from(42u64))); + fn install_is_capped() { + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(4); + for _ in 0..MAX_FILTERS { + manager.install_block(tx.subscribe()).expect("under the cap"); + } + assert_eq!( + manager.install_block(tx.subscribe()), + Err(TooManyFilters), + "installing beyond the cap is rejected" + ); } #[test] - fn uninstall_removes_filter() { - let manager = manager(); - let id = manager.install_logs(Filter::new(), 0, None); - assert!(manager.uninstall(id)); - assert!(!manager.uninstall(id), "a filter can only be uninstalled once"); - assert!(manager.poll(id).is_none(), "a polled-after-uninstall filter is gone"); + fn expired_filter_is_evicted_on_poll() { + let manager = FilterManager::new(); + let (tx, _rx) = broadcast::channel::(4); + let id = manager.install_block(tx.subscribe()).unwrap(); + expire(&manager, id); + assert!(manager.poll_changes(id).is_none(), "an expired filter should not be polled"); + assert!(!manager.uninstall(id), "an expired filter should already be evicted"); } #[test] - fn expired_filter_is_evicted_on_poll() { - let manager = - FilterManager { inner: Arc::new(Mutex::new(HashMap::new())), timeout: Duration::ZERO }; - let id = manager.install_block(0); - // With a zero timeout any elapsed time counts as expired. - std::thread::sleep(Duration::from_millis(1)); - assert!(manager.poll(id).is_none(), "an expired filter should not be polled"); - assert!(!manager.uninstall(id), "an expired filter should already be evicted"); + fn cleanup_drops_expired_and_stale_filters() { + let manager = FilterManager::new(); + let (blocks, _b) = broadcast::channel::(4); + let (logs, _l) = broadcast::channel::(4); + + // Fresh block filter: kept. + let fresh = manager.install_block(blocks.subscribe()).unwrap(); + // Expired filter: dropped on the deadline alone. + let stale_deadline = manager.install_block(blocks.subscribe()).unwrap(); + expire(&manager, stale_deadline); + // Log filter bounded to block 5: dropped once the head passes it, even though fresh. + let bounded = manager.install_logs(Filter::new().to_block(5u64), logs.subscribe()).unwrap(); + + manager.evict(Some(10)); + + assert!(manager.poll_changes(fresh).is_some(), "fresh filter survives"); + assert!(manager.poll_changes(stale_deadline).is_none(), "expired filter dropped"); + assert!(manager.poll_changes(bounded).is_none(), "filter past its toBlock dropped"); } } diff --git a/substrate/frame/revive/rpc/src/lib.rs b/substrate/frame/revive/rpc/src/lib.rs index ad61555aa1db..01f87a767518 100644 --- a/substrate/frame/revive/rpc/src/lib.rs +++ b/substrate/frame/revive/rpc/src/lib.rs @@ -18,7 +18,7 @@ #![cfg_attr(docsrs, feature(doc_cfg))] pub use alloy_rpc_types::{BlockId, BlockNumberOrTag, Filter, FilterBlockOption}; -use client::{ClientError, SubstrateBlockNumber}; +use client::ClientError; use futures::{Stream, StreamExt, TryStreamExt}; use jsonrpsee::{ PendingSubscriptionSink, SubscriptionMessage, SubscriptionSink, @@ -83,8 +83,11 @@ pub struct EthRpcServerImpl { /// When true, estimate_gas uses Pending block if no block is specified. use_pending_for_estimate_gas: bool, - /// Registry of installed polling filters (`eth_newFilter` / `eth_newBlockFilter`). - filters: FilterManager, + /// Registry backing the polling filter API (`eth_newFilter`, `eth_newBlockFilter`, + /// `eth_getFilterChanges`, `eth_getFilterLogs`, `eth_uninstallFilter`). It drains the same + /// block and log broadcast channels that `eth_subscribe` consumes, so the two APIs share one + /// event source. + filter_registry: FilterManager, } impl EthRpcServerImpl { @@ -95,7 +98,7 @@ impl EthRpcServerImpl { accounts: vec![], allow_unprotected_txs: false, use_pending_for_estimate_gas: false, - filters: FilterManager::new(), + filter_registry: FilterManager::new(), } } @@ -116,6 +119,12 @@ impl EthRpcServerImpl { self.use_pending_for_estimate_gas = use_pending_for_estimate_gas; self } + + /// Sets the filter registry, so the caller can share it with its background cleanup task. + pub fn with_filter_registry(mut self, filter_registry: FilterManager) -> Self { + self.filter_registry = filter_registry; + self + } } /// The error type for the EVM RPC server. @@ -145,6 +154,9 @@ pub enum EthRpcError { /// No filter is registered under the given id (it is unknown or has expired). #[error("filter not found")] FilterNotFound, + /// The maximum number of concurrently installed filters has been reached. + #[error("filter limit reached")] + TooManyFilters, } impl From for ErrorObjectOwned { @@ -161,6 +173,7 @@ impl From for ErrorObjectOwned { EthRpcError::AccountNotFound(_) | EthRpcError::InvalidTransaction | EthRpcError::FilterNotFound | + EthRpcError::TooManyFilters | EthRpcError::TransactionTypeNotSupported(_) => CALL_EXECUTION_FAILED_CODE, }; Self::owned::(code, message, None) @@ -445,67 +458,44 @@ impl EthRpcServer for EthRpcServerImpl { } async fn new_filter(&self, filter: Filter) -> RpcResult { - let head = self.client.block_number().await?; - let (from, to) = self.resolve_filter_range(&filter, head).await?; - Ok(self.filters.install_logs(filter, from, to)) + // Subscribe to the shared log channel now, so the filter reports logs from blocks imported + // after installation, matching go-ethereum. Its `fromBlock`/`toBlock` only bound the + // historical replay served by `eth_getFilterLogs`. + let id = self + .filter_registry + .install_logs(filter, self.client.get_log_subscription_rx()) + .map_err(|_| EthRpcError::TooManyFilters)?; + Ok(id.into()) } async fn new_block_filter(&self) -> RpcResult { - let head = self.client.block_number().await?; - // Report only blocks added after the filter is installed, matching go-ethereum. - Ok(self.filters.install_block(head.saturating_add(1))) + // The receiver starts at the current channel head, so only blocks imported after this point + // are reported — no manual head bookkeeping. + let id = self + .filter_registry + .install_block(self.client.get_block_subscription_rx()) + .map_err(|_| EthRpcError::TooManyFilters)?; + Ok(id.into()) } async fn get_filter_changes(&self, filter_id: U256) -> RpcResult { - match self.filters.poll(filter_id).ok_or(EthRpcError::FilterNotFound)? { - FilterPoll::Logs { mut filter, from, to } => { - let head = self.client.block_number().await?; - let upper = to.unwrap_or(head).min(head); - if from > upper { - return Ok(FilterResults::Logs(vec![])); - } - - filter.block_option = FilterBlockOption::Range { - from_block: Some(BlockNumberOrTag::Number(from)), - to_block: Some(BlockNumberOrTag::Number(upper)), - }; - let logs = self.client.logs(Some(filter)).await?; - self.filters.advance(filter_id, upper.saturating_add(1)); - Ok(FilterResults::Logs(logs)) - }, - FilterPoll::Block { from } => { - let head = self.client.block_number().await?; - if from > head { - return Ok(FilterResults::Hashes(vec![])); - } - - let mut hashes = Vec::new(); - for number in from..=head { - let Some(block) = self - .client - .block_by_number_or_tag(&BlockNumberOrTag::Number(number)) - .await? - else { - continue; - }; - if let Some(eth_block) = self.client.evm_block(block, false).await { - hashes.push(eth_block.hash); - } - } - self.filters.advance(filter_id, head.saturating_add(1)); - Ok(FilterResults::Hashes(hashes)) - }, - } + self.filter_registry + .poll_changes(filter_id.into()) + .ok_or_else(|| EthRpcError::FilterNotFound.into()) } - async fn get_filter_logs(&self, filter_id: U256) -> RpcResult { - let filter = self.filters.logs_filter(filter_id).ok_or(EthRpcError::FilterNotFound)?; - let logs = self.client.logs(Some(filter)).await?; - Ok(FilterResults::Logs(logs)) + async fn get_filter_logs(&self, filter_id: U256) -> RpcResult> { + // `eth_getFilterLogs` replays the filter's whole range, exactly as `eth_getLogs` would, so + // it reuses the same range resolution rather than the streamed changes. + let filter = self + .filter_registry + .logs_filter(filter_id.into()) + .ok_or(EthRpcError::FilterNotFound)?; + Ok(self.client.logs(Some(filter)).await?) } async fn uninstall_filter(&self, filter_id: U256) -> RpcResult { - Ok(self.filters.uninstall(filter_id)) + Ok(self.filter_registry.uninstall(filter_id.into())) } async fn get_storage_at( @@ -634,69 +624,6 @@ impl EthRpcServer for EthRpcServerImpl { } impl EthRpcServerImpl { - /// Resolve the block range of a newly installed log filter into concrete block numbers. - /// - /// Returns `(from, to)` where `from` is the first block a poll should report and `to` is the - /// inclusive upper bound (`None` to follow the chain head). A filter without an explicit - /// `fromBlock`, or one pinned to `latest`/`pending`, reports only blocks added after it was - /// installed, matching go-ethereum. - async fn resolve_filter_range( - &self, - filter: &Filter, - head: SubstrateBlockNumber, - ) -> RpcResult<(SubstrateBlockNumber, Option)> { - match &filter.block_option { - FilterBlockOption::AtBlockHash(block_hash) => { - let number = self - .resolve_ethereum_block_number(H256::from_slice(block_hash.as_slice())) - .await?; - Ok((number, Some(number))) - }, - FilterBlockOption::Range { from_block, to_block } => { - let from = match from_block { - None | Some(BlockNumberOrTag::Latest) | Some(BlockNumberOrTag::Pending) => { - head.saturating_add(1) - }, - Some(tag) => self.resolve_block_tag(tag, head).await?, - }; - let to = match to_block { - None | Some(BlockNumberOrTag::Latest) | Some(BlockNumberOrTag::Pending) => None, - Some(tag) => Some(self.resolve_block_tag(tag, head).await?), - }; - Ok((from, to)) - }, - } - } - - /// Resolve a block tag (a number, or `earliest`/`finalized`/`safe`) into a concrete block - /// number, falling back to the chain head if the tagged block cannot be located. - async fn resolve_block_tag( - &self, - tag: &BlockNumberOrTag, - head: SubstrateBlockNumber, - ) -> RpcResult { - if let BlockNumberOrTag::Number(number) = tag { - return Ok(*number); - } - match self.client.block_by_number_or_tag(tag).await? { - Some(block) => Ok(block.number()), - None => Ok(head), - } - } - - /// Resolve an Ethereum block hash to its block number, erroring if the block is unknown. - async fn resolve_ethereum_block_number( - &self, - block_hash: H256, - ) -> RpcResult { - let block = self - .client - .block_by_ethereum_hash(&block_hash) - .await? - .ok_or(ClientError::BlockNotFound)?; - Ok(block.number()) - } - async fn get_transaction_by_substrate_block_hash_and_index( &self, substrate_block_hash: H256, @@ -772,6 +699,7 @@ mod error_codes_tests { CALL_EXECUTION_FAILED_CODE, ), (EthRpcError::FilterNotFound, CALL_EXECUTION_FAILED_CODE), + (EthRpcError::TooManyFilters, CALL_EXECUTION_FAILED_CODE), ]; for (err, expected_code) in cases { diff --git a/substrate/frame/revive/rpc/src/tests.rs b/substrate/frame/revive/rpc/src/tests.rs index d7cf238596ed..663ad7fcc0c5 100644 --- a/substrate/frame/revive/rpc/src/tests.rs +++ b/substrate/frame/revive/rpc/src/tests.rs @@ -691,8 +691,50 @@ fn is_empty_filter_results(results: &FilterResults) -> bool { } } +/// Poll `eth_getFilterChanges` until it yields at least one log, or fail after a bounded wait: a +/// streamed log may not have reached the filter the instant its receipt becomes available. +async fn poll_until_logs( + client: &C, + filter_id: U256, +) -> anyhow::Result> { + for _ in 0..40 { + match client.get_filter_changes(filter_id).await? { + FilterResults::Logs(logs) if !logs.is_empty() => return Ok(logs), + FilterResults::Logs(_) => {}, + other => { + return Err(anyhow!("Expected Logs from eth_getFilterChanges, got: {other:?}")); + }, + } + tokio::time::sleep(tokio::time::Duration::from_millis(250)).await; + } + Err(anyhow!("filter reported no logs within the timeout")) +} + +/// Poll `eth_getFilterChanges` until it yields at least one block hash, or fail after a bounded +/// wait. +async fn poll_until_hashes( + client: &C, + filter_id: U256, +) -> anyhow::Result> { + for _ in 0..40 { + match client.get_filter_changes(filter_id).await? { + FilterResults::Hashes(hashes) if !hashes.is_empty() => return Ok(hashes), + FilterResults::Hashes(_) => {}, + other => { + return Err(anyhow!("Expected Hashes from eth_getFilterChanges, got: {other:?}")); + }, + } + tokio::time::sleep(tokio::time::Duration::from_millis(250)).await; + } + Err(anyhow!("block filter reported no hashes within the timeout")) +} + /// Exercise the polling filter API end to end: `eth_newFilter`, `eth_getFilterChanges`, /// `eth_getFilterLogs`, `eth_newBlockFilter` and `eth_uninstallFilter`. +/// +/// The polling API and `eth_subscribe` are fed by the same block/log broadcast channels, so a log +/// filter reports only logs from blocks imported after it was installed; its `fromBlock`/`toBlock` +/// bound only the historical `eth_getFilterLogs` replay. async fn test_filter_lifecycle() -> anyhow::Result<()> { let client = Arc::new(SharedResources::client().await); let account = Account::default(); @@ -708,6 +750,13 @@ async fn test_filter_lifecycle() -> anyhow::Result<()> { let contract_address = create1(&account.address(), nonce.try_into().unwrap()); assert_eq!(Some(contract_address), receipt.contract_address); + // Install a log filter scoped to this (uniquely addressed) contract *before* emitting, so the + // event streams to it. `fromBlock` is earliest so `eth_getFilterLogs` replays its history. + let filter = Filter::new() + .from_block(BlockNumberOrTag::Earliest) + .address(AlloyAddress::from_slice(contract_address.as_ref())); + let filter_id = client.new_filter(filter).await?; + // Emit a single event. let value = U256::from(1_000_000_000_000u128); let tx = TransactionBuilder::new(client.clone()) @@ -717,17 +766,8 @@ async fn test_filter_lifecycle() -> anyhow::Result<()> { .await?; let emit_receipt = tx.wait_for_receipt().await?; - // A log filter scoped to this (uniquely addressed) contract, starting at the emit block. - let filter = Filter::new() - .from_block(emit_receipt.block_number.as_u64()) - .address(AlloyAddress::from_slice(contract_address.as_ref())); - let filter_id = client.new_filter(filter).await?; - - // The first poll returns the emitted log. - let logs = match client.get_filter_changes(filter_id).await? { - FilterResults::Logs(logs) => logs, - other => return Err(anyhow!("Expected Logs from eth_getFilterChanges, got: {other:?}")), - }; + // The filter streams the freshly emitted log. + let logs = poll_until_logs(client.as_ref(), filter_id).await?; assert_eq!(logs.len(), 1, "filter should report exactly the one emitted log"); let event_signature = H256(sp_io::hashing::keccak_256(b"Received(address,uint256)")); assert_eq!(logs[0].address, contract_address); @@ -735,14 +775,18 @@ async fn test_filter_lifecycle() -> anyhow::Result<()> { assert_eq!(logs[0].topics[0], event_signature); assert_eq!(logs[0].transaction_hash, emit_receipt.transaction_hash); - // `eth_getFilterLogs` replays the full range and returns the same log. - let all_logs = match client.get_filter_logs(filter_id).await? { - FilterResults::Logs(logs) => logs, - other => return Err(anyhow!("Expected Logs from eth_getFilterLogs, got: {other:?}")), - }; - assert_eq!(all_logs, logs, "getFilterLogs should replay the same log"); + // `eth_getFilterLogs` replays the contract's history and returns a plain vector of logs. + let all_logs = client.get_filter_logs(filter_id).await?; + assert!( + all_logs.iter().any(|log| log.transaction_hash == emit_receipt.transaction_hash), + "getFilterLogs should include the emitted log" + ); + assert!( + all_logs.iter().all(|log| log.address == contract_address), + "getFilterLogs should only return logs from the filtered address" + ); - // A second poll has no new logs, as the cursor advanced past the emit block. + // A second poll has no new logs, since the previous changes were already drained. let empty = client.get_filter_changes(filter_id).await?; assert!(is_empty_filter_results(&empty), "a second poll should be empty, got: {empty:?}"); @@ -761,7 +805,7 @@ async fn test_filter_lifecycle() -> anyhow::Result<()> { }; assert_eq!(call.message(), "filter not found"); - // A block filter reports the hashes of blocks added after it was installed. + // A block filter reports the hashes of blocks imported after it was installed. let block_filter_id = client.new_block_filter().await?; let tx = TransactionBuilder::new(client.clone()) .value(value) @@ -769,12 +813,8 @@ async fn test_filter_lifecycle() -> anyhow::Result<()> { .send() .await?; tx.wait_for_receipt().await?; - let block_hashes = match client.get_filter_changes(block_filter_id).await? { - FilterResults::Hashes(hashes) => hashes, - other => return Err(anyhow!("Expected Hashes from a block filter, got: {other:?}")), - }; - assert!(!block_hashes.is_empty(), "block filter should report the newly produced block"); - let newest = block_hashes.last().expect("non-empty checked above"); + let block_hashes = poll_until_hashes(client.as_ref(), block_filter_id).await?; + let newest = block_hashes.last().expect("non-empty checked in the poll helper"); assert!( client.get_block_by_hash(*newest, false).await?.is_some(), "a reported block hash should resolve to a real block" diff --git a/substrate/frame/revive/rpc/src/types/subscription.rs b/substrate/frame/revive/rpc/src/types/subscription.rs index d7d33637f34d..4dca717fc3ce 100644 --- a/substrate/frame/revive/rpc/src/types/subscription.rs +++ b/substrate/frame/revive/rpc/src/types/subscription.rs @@ -139,6 +139,29 @@ impl LogsSubscriptionFilter { } } + /// Builds a matcher from an `eth_newFilter` [`Filter`], so a polling log filter selects + /// streamed logs with the same address/topics logic used for `eth_subscribe`. + pub fn from_filter(filter: &Filter) -> Self { + let addresses = (!filter.address.is_empty()).then(|| { + filter + .address + .iter() + .map(|address| H160::from_slice(address.as_slice())) + .collect() + }); + let topics = filter.topics.iter().any(|topic| !topic.is_empty()).then(|| { + let mut resolved_topics: [Option>; 4] = [None, None, None, None]; + for (index, topic) in filter.topics.iter().enumerate() { + if !topic.is_empty() { + resolved_topics[index] = + Some(topic.iter().map(|hash| H256::from_slice(hash.as_slice())).collect()); + } + } + resolved_topics + }); + Self { addresses, topics } + } + /// Checks if a certain log matches this filter. pub fn matches(&self, log: &Log) -> bool { // Check the emitter address. If it doesn't match, then we return.