Implement Filecoin.EthGetFilterChanges (#5383)

Author: sudo-shashank

CHANGELOG.md

@@ -33,6 +33,8 @@
 
 - [#5386](https://github.com/ChainSafe/forest/pull/5386) Add support for the `Filecoin.EthTraceTransaction` RPC method.
 
+- [#5383](https://github.com/ChainSafe/forest/pull/5383) Add support for `Filecoin.EthGetFilterChanges` RPC method.
+
 ### Changed
 
 ### Removed

src/rpc/methods/eth.rs

@@ -24,7 +24,9 @@ use crate::interpreter::VMTrace;
 use crate::lotus_json::{lotus_json_with_self, HasLotusJson};
 use crate::message::{ChainMessage, Message as _, SignedMessage};
 use crate::rpc::error::ServerError;
-use crate::rpc::eth::filter::{event::EventFilter, SkipEvent};
+use crate::rpc::eth::filter::{
+    event::EventFilter, mempool::MempoolFilter, tipset::TipSetFilter, SkipEvent,
+};
 use crate::rpc::eth::types::{EthBlockTrace, EthTrace};
 use crate::rpc::types::{ApiTipsetKey, EventEntry, MessageLookup};
 use crate::rpc::EthEventHandler;
@@ -50,13 +52,15 @@ use crate::utils::encoding::from_slice_with_fallback;
 use crate::utils::multihash::prelude::*;
 use anyhow::{anyhow, bail, Context, Error, Result};
 use cid::Cid;
+use filter::{ParsedFilter, ParsedFilterTipsets};
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_encoding::{RawBytes, CBOR, DAG_CBOR, IPLD_RAW};
 use ipld_core::ipld::Ipld;
 use itertools::Itertools;
 use num::{BigInt, Zero as _};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use std::ops::RangeInclusive;
 use std::str::FromStr;
 use std::sync::Arc;
 use utils::{decode_payload, lookup_eth_address};
@@ -2730,6 +2734,36 @@ where
         .collect()
 }
 
+fn eth_filter_logs_from_tipsets(events: &[CollectedEvent]) -> anyhow::Result<Vec<EthHash>> {
+    events
+        .iter()
+        .map(|event| event.tipset_key.cid().map(Into::into))
+        .collect()
+}
+
+fn eth_filter_logs_from_messages<DB: Blockstore>(
+    ctx: &Ctx<DB>,
+    events: &[CollectedEvent],
+) -> anyhow::Result<Vec<EthHash>> {
+    events
+        .iter()
+        .filter_map(|event| {
+            match eth_tx_hash_from_message_cid(
+                ctx.store(),
+                &event.msg_cid,
+                ctx.state_manager.chain_config().eth_chain_id,
+            ) {
+                Ok(Some(hash)) => Some(Ok(hash)),
+                Ok(None) => {
+                    tracing::warn!("Ignoring event");
+                    None
+                }
+                Err(err) => Some(Err(err)),
+            }
+        })
+        .collect()
+}
+
 fn eth_filter_logs_from_events<DB: Blockstore>(
     ctx: &Ctx<DB>,
     events: &[CollectedEvent],
@@ -2775,6 +2809,19 @@ fn eth_filter_result_from_events<DB: Blockstore>(
     )?))
 }
 
+fn eth_filter_result_from_tipsets(events: &[CollectedEvent]) -> anyhow::Result<EthFilterResult> {
+    Ok(EthFilterResult::Txs(eth_filter_logs_from_tipsets(events)?))
+}
+
+fn eth_filter_result_from_messages<DB: Blockstore>(
+    ctx: &Ctx<DB>,
+    events: &[CollectedEvent],
+) -> anyhow::Result<EthFilterResult> {
+    Ok(EthFilterResult::Txs(eth_filter_logs_from_messages(
+        ctx, events,
+    )?))
+}
+
 pub enum EthGetLogs {}
 impl RpcMethod<1> for EthGetLogs {
     const NAME: &'static str = "Filecoin.EthGetLogs";
@@ -2847,6 +2894,120 @@ impl RpcMethod<1> for EthGetFilterLogs {
     }
 }
 
+pub enum EthGetFilterChanges {}
+impl RpcMethod<1> for EthGetFilterChanges {
+    const NAME: &'static str = "Filecoin.EthGetFilterChanges";
+    const NAME_ALIAS: Option<&'static str> = Some("eth_getFilterChanges");
+    const N_REQUIRED_PARAMS: usize = 1;
+    const PARAM_NAMES: [&'static str; 1] = ["filterId"];
+    const API_PATHS: ApiPaths = ApiPaths::V1;
+    const PERMISSION: Permission = Permission::Write;
+    const DESCRIPTION: Option<&'static str> =
+        Some("Returns event logs which occured since the last poll");
+
+    type Params = (FilterID,);
+    type Ok = EthFilterResult;
+    async fn handle(
+        ctx: Ctx<impl Blockstore + Send + Sync + 'static>,
+        (filter_id,): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        let eth_event_handler = ctx.eth_event_handler.clone();
+        if let Some(store) = &eth_event_handler.filter_store {
+            let filter = store.get(&filter_id)?;
+            if let Some(event_filter) = filter.as_any().downcast_ref::<EventFilter>() {
+                let events = ctx
+                    .eth_event_handler
+                    .get_events_for_parsed_filter(
+                        &ctx,
+                        &event_filter.into(),
+                        SkipEvent::OnUnresolvedAddress,
+                    )
+                    .await?;
+                let recent_events: Vec<CollectedEvent> = events
+                    .clone()
+                    .into_iter()
+                    .filter(|event| !event_filter.collected.contains(event))
+                    .collect();
+                let filter = Arc::new(EventFilter {
+                    id: event_filter.id.clone(),
+                    tipsets: event_filter.tipsets.clone(),
+                    addresses: event_filter.addresses.clone(),
+                    keys_with_codec: event_filter.keys_with_codec.clone(),
+                    max_results: event_filter.max_results,
+                    collected: events.clone(),
+                });
+                store.update(filter);
+                return Ok(eth_filter_result_from_events(&ctx, &recent_events)?);
+            }
+            if let Some(tipset_filter) = filter.as_any().downcast_ref::<TipSetFilter>() {
+                let events = ctx
+                    .eth_event_handler
+                    .get_events_for_parsed_filter(
+                        &ctx,
+                        &ParsedFilter::new_with_tipset(ParsedFilterTipsets::Range(
+                            // heaviest tipset doesn't have events because its messages haven't been executed yet
+                            RangeInclusive::new(
+                                tipset_filter
+                                    .collected
+                                    .unwrap_or(ctx.chain_store().heaviest_tipset().epoch() - 1),
+                                // Use -1 to indicate that the range extends until the latest available tipset.
+                                -1,
+                            ),
+                        )),
+                        SkipEvent::OnUnresolvedAddress,
+                    )
+                    .await?;
+                let new_collected = events
+                    .iter()
+                    .max_by_key(|event| event.height)
+                    .map(|e| e.height);
+                if let Some(height) = new_collected {
+                    let filter = Arc::new(TipSetFilter {
+                        id: tipset_filter.id.clone(),
+                        max_results: tipset_filter.max_results,
+                        collected: Some(height),
+                    });
+                    store.update(filter);
+                }
+                return Ok(eth_filter_result_from_tipsets(&events)?);
+            }
+            if let Some(mempool_filter) = filter.as_any().downcast_ref::<MempoolFilter>() {
+                let events = ctx
+                    .eth_event_handler
+                    .get_events_for_parsed_filter(
+                        &ctx,
+                        &ParsedFilter::new_with_tipset(ParsedFilterTipsets::Range(
+                            // heaviest tipset doesn't have events because its messages haven't been executed yet
+                            RangeInclusive::new(
+                                mempool_filter
+                                    .collected
+                                    .unwrap_or(ctx.chain_store().heaviest_tipset().epoch() - 1),
+                                // Use -1 to indicate that the range extends until the latest available tipset.
+                                -1,
+                            ),
+                        )),
+                        SkipEvent::OnUnresolvedAddress,
+                    )
+                    .await?;
+                let new_collected = events
+                    .iter()
+                    .max_by_key(|event| event.height)
+                    .map(|e| e.height);
+                if let Some(height) = new_collected {
+                    let filter = Arc::new(MempoolFilter {
+                        id: mempool_filter.id.clone(),
+                        max_results: mempool_filter.max_results,
+                        collected: Some(height),
+                    });
+                    store.update(filter);
+                }
+                return Ok(eth_filter_result_from_messages(&ctx, &events)?);
+            }
+        }
+        Err(anyhow::anyhow!("method not supported").into())
+    }
+}
+
 pub enum EthTraceBlock {}
 impl RpcMethod<1> for EthTraceBlock {
     const NAME: &'static str = "Filecoin.EthTraceBlock";

src/rpc/methods/eth/filter/event.rs

@@ -13,11 +13,17 @@ use std::any::Any;
 #[allow(dead_code)]
 #[derive(Debug, PartialEq)]
 pub struct EventFilter {
+    // Unique id used to identify the filter
     pub id: FilterID,
+    // Tipsets to filter
     pub tipsets: ParsedFilterTipsets,
-    pub addresses: Vec<Address>, // list of actor addresses that are extpected to emit the event
-    pub keys_with_codec: HashMap<String, Vec<ActorEventBlock>>, // map of key names to a list of alternate values that may match
-    pub max_results: usize, // maximum number of results to collect
+    // list of actor addresses that are extpected to emit the event
+    pub addresses: Vec<Address>,
+    // Map of key names to a list of alternate values that may match
+    pub keys_with_codec: HashMap<String, Vec<ActorEventBlock>>,
+    // Maximum number of results to collect
+    pub max_results: usize,
+    // Collected events
     pub collected: Vec<CollectedEvent>,
 }
 

src/rpc/methods/eth/filter/mempool.rs

@@ -3,6 +3,7 @@
 
 use crate::rpc::eth::{filter::Filter, filter::FilterManager, FilterID};
 use crate::rpc::Arc;
+use crate::shim::fvm_shared_latest::clock::ChainEpoch;
 use ahash::AHashMap as HashMap;
 use anyhow::{Context, Result};
 use parking_lot::RwLock;
@@ -13,14 +14,22 @@ use std::any::Any;
 #[allow(dead_code)]
 #[derive(Debug, PartialEq)]
 pub struct MempoolFilter {
-    id: FilterID,       // Unique id used to identify the filter
-    max_results: usize, // maximum number of results to collect
+    // Unique id used to identify the filter
+    pub id: FilterID,
+    // Maximum number of results to collect
+    pub max_results: usize,
+    // Epoch at which the results were collected
+    pub collected: Option<ChainEpoch>,
 }
 
 impl MempoolFilter {
     pub fn new(max_results: usize) -> Result<Arc<Self>, uuid::Error> {
         let id = FilterID::new()?;
-        Ok(Arc::new(Self { id, max_results }))
+        Ok(Arc::new(Self {
+            id,
+            max_results,
+            collected: None,
+        }))
     }
 }
 

src/rpc/methods/eth/filter/mod.rs

@@ -15,9 +15,9 @@
 //! - **TipSet Filter**: Tracks changes in the blockchain's tipset (the latest set of blocks).
 //! - **Mempool Filter**: Monitors the Ethereum mempool for new pending transactions that meet certain criteria.
 pub mod event;
-mod mempool;
+pub mod mempool;
 mod store;
-mod tipset;
+pub mod tipset;
 
 use super::get_tipset_from_hash;
 use super::BlockNumberOrHash;
@@ -643,6 +643,13 @@ pub struct ParsedFilter {
 }
 
 impl ParsedFilter {
+    pub fn new_with_tipset(tipsets: ParsedFilterTipsets) -> Self {
+        ParsedFilter {
+            tipsets,
+            addresses: vec![],
+            keys: HashMap::new(),
+        }
+    }
     pub fn from_actor_event_filter(
         chain_height: ChainEpoch,
         _max_filter_height_range: ChainEpoch,

src/rpc/methods/eth/filter/tipset.rs

@@ -3,6 +3,7 @@
 
 use crate::rpc::eth::{filter::Filter, filter::FilterManager, FilterID};
 use crate::rpc::Arc;
+use crate::shim::fvm_shared_latest::clock::ChainEpoch;
 use ahash::AHashMap as HashMap;
 use anyhow::{Context, Result};
 use parking_lot::RwLock;
@@ -11,14 +12,22 @@ use std::any::Any;
 #[allow(dead_code)]
 #[derive(Debug, PartialEq)]
 pub struct TipSetFilter {
-    id: FilterID,
-    max_results: usize,
+    // Unique id used to identify the filter
+    pub id: FilterID,
+    // Maximum number of results to collect
+    pub max_results: usize,
+    // Epoch at which the results were collected
+    pub collected: Option<ChainEpoch>,
 }
 
 impl TipSetFilter {
     pub fn new(max_results: usize) -> Result<Arc<Self>, uuid::Error> {
         let id = FilterID::new()?;
-        Ok(Arc::new(Self { id, max_results }))
+        Ok(Arc::new(Self {
+            id,
+            max_results,
+            collected: None,
+        }))
     }
 }
 

src/rpc/mod.rs

@@ -92,6 +92,7 @@ macro_rules! for_each_rpc_method {
         $callback!($crate::rpc::eth::EthGetCode);
         $callback!($crate::rpc::eth::EthGetLogs);
         $callback!($crate::rpc::eth::EthGetFilterLogs);
+        $callback!($crate::rpc::eth::EthGetFilterChanges);
         $callback!($crate::rpc::eth::EthGetMessageCidByTransactionHash);
         $callback!($crate::rpc::eth::EthGetStorageAt);
         $callback!($crate::rpc::eth::EthGetTransactionByHash);

src/tool/subcommands/api_cmd/api_compare_tests.rs

@@ -1643,6 +1643,8 @@ fn eth_tests_with_tipset<DB: Blockstore>(store: &Arc<DB>, shared_tipset: &Tipset
         .sort_policy(SortPolicy::All),
         RpcTest::identity(EthGetFilterLogs::request((FilterID::new().unwrap(),)).unwrap())
             .policy_on_rejected(PolicyOnRejected::PassWithIdenticalError),
+        RpcTest::identity(EthGetFilterChanges::request((FilterID::new().unwrap(),)).unwrap())
+            .policy_on_rejected(PolicyOnRejected::PassWithIdenticalError),
         RpcTest::identity(EthGetTransactionHashByCid::request((block_cid,)).unwrap()),
         RpcTest::identity(
             EthTraceBlock::request((ExtBlockNumberOrHash::from_block_number(