feat(filters): add ttl in eth filters (#7386)

Author: akaladarshi

CHANGELOG.md

@@ -51,6 +51,8 @@ Non-mandatory release for all node operators. It includes some fixes and improve
 
 - [#7376](https://github.com/ChainSafe/forest/pull/7376): Added `--augmented-snapshot` and `--tipset-lookup` to `forest-cli snapshot export`.
 
+- [#7265](https://github.com/ChainSafe/forest/issues/7265): Expire installed `eth` filters that have not been polled for longer than a TTL. The TTL defaults to one hour (Lotus parity) and is configurable via `filter_ttl_secs` in the `[events]` config section or the `FOREST_FILTER_TTL_SECS` environment variable, `0` disables expiry.
+
 ### Changed
 
 - [#7405](https://github.com/ChainSafe/forest/pull/7405): `Filecoin.StateReplay` is now served from the shared tipset trace cache, so replaying every transaction of a block executes the tipset once instead of once per call. Tipset replays triggered by RPC (`StateReplay`, `trace_*`, `debug_trace*`) are now bounded to half the available CPUs by default, configurable via `FOREST_RPC_REPLAY_CONCURRENCY`.

docs/docs/users/reference/env_variables.md

@@ -30,6 +30,7 @@ process.
 | `FOREST_MAX_FILTERS`                                             | integer                          | 100                                                        | 100                                                           | The maximum number of filters                                                                                                                                                                                                                                                  |
 | `FOREST_MAX_FILTER_RESULTS`                                      | positive integer                 | 10,000                                                     | 10000                                                         | The maximum number of filter results                                                                                                                                                                                                                                           |
 | `FOREST_MAX_FILTER_HEIGHT_RANGE`                                 | positive integer                 | 2880                                                       | 2880                                                          | The maximum filter height range allowed, a conservative limit of one day                                                                                                                                                                                                       |
+| `FOREST_FILTER_TTL_SECS`                                         | integer                          | 3600                                                       | 300                                                           | Installed `eth` filters that have not been polled for longer than this many seconds are removed. Set to `0` to never expire filters                                                                                                                                            |
 | `FOREST_STATE_MIGRATION_THREADS`                                 | integer                          | Depends on the machine.                                    | 3                                                             | The number of threads for state migration thread-pool. Advanced users only.                                                                                                                                                                                                    |
 | `FOREST_CONFIG_PATH`                                             | string                           | /$FOREST_HOME/com.ChainSafe.Forest/config.toml             | `/path/to/config.toml`                                        | Forest configuration path. Alternatively supplied via `--config` cli parameter.                                                                                                                                                                                                |
 | `FOREST_PATH`                                                    | directory path                   | platform data directory (e.g. `$HOME/.local/share/forest`) | `/var/lib/forest`                                             | Override the Forest data directory. Takes precedence over the `data_dir` set in the configuration file and the default. Honored by all `forest*` binaries, so `forest-cli`/`forest-tool` also read the JWT admin token from this directory. Equivalent to Lotus' `LOTUS_PATH`. |

src/cli_shared/cli/config.rs

@@ -43,6 +43,7 @@ impl Default for DaemonConfig {
 /// Structure that defines events configuration
 #[derive(Deserialize, Serialize, PartialEq, Eq, Debug, Clone)]
 #[cfg_attr(test, derive(derive_quickcheck_arbitrary::Arbitrary))]
+#[serde(default)]
 pub struct EventsConfig {
     /// Caps the events returned by event-filter queries used by the actor
     /// events API and the Ethereum event and receipt APIs (`eth_getLogs`,
@@ -60,13 +61,17 @@ pub struct EventsConfig {
     pub max_filter_results: usize,
     /// Maximum block-range span (in epochs) accepted in event-filter queries.
     pub max_filter_height_range: ChainEpoch,
+    /// Installed `eth` filters idle for longer than this many
+    /// seconds are removed. Set to `0` to never expire filters.
+    pub filter_ttl_secs: u64,
 }
 
 impl Default for EventsConfig {
     fn default() -> Self {
         Self {
             max_filter_results: 10000,
             max_filter_height_range: 2880,
+            filter_ttl_secs: 3600,
         }
     }
 }

src/daemon/mod.rs

@@ -559,6 +559,12 @@ fn maybe_start_rpc_service(
             ctx.chain_config().eth_chain_id,
             mpool.subscriber(),
         ));
+        if let Some(ttl) = eth_event_handler.filter_ttl() {
+            services.spawn({
+                let eth_event_handler = eth_event_handler.clone();
+                async move { eth_event_handler.run_filter_gc(ttl).await }
+            });
+        }
         if is_env_truthy("FOREST_JWT_DISABLE_EXP_VALIDATION") {
             warn!(
                 "JWT expiration validation is disabled; this significantly weakens security and should only be used in tightly controlled environments"

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

@@ -52,9 +52,15 @@ use fvm_ipld_encoding::IPLD_RAW;
 use serde::*;
 use std::ops::RangeInclusive;
 use std::sync::Arc;
+use std::time::Duration;
 use store::*;
 use tokio_util::task::AbortOnDropHandle;
 
+/// Longest period between expired-filter sweeps; a TTL shorter than this is
+/// swept every TTL instead. Mirrors Lotus's hardcoded 30-minute GC ticker:
+/// <https://github.com/filecoin-project/lotus/blob/release/v1.36.1/node/impl/eth/events.go#L361>
+const FILTER_GC_INTERVAL: Duration = Duration::from_mins(30);
+
 /// A trait for filtering events based on predefined conditions.
 ///
 /// Implementors of this trait define custom logic to determine whether an event matches the filtering criteria
@@ -120,6 +126,7 @@ pub struct EthEventHandler {
     pub filter_store: Option<Arc<dyn FilterStore>>,
     pub max_filter_results: usize,
     pub max_filter_height_range: ChainEpoch,
+    filter_ttl: Option<Duration>,
     event_filter_manager: Option<Arc<EventFilterManager>>,
     tipset_filter_manager: Option<Arc<TipSetFilterManager>>,
     mempool_filter_manager: Option<Arc<MempoolFilterManager>>,
@@ -173,6 +180,8 @@ impl EthEventHandler {
                 }
             })
             .unwrap_or(config.max_filter_height_range);
+        let filter_ttl_secs: u64 = env_or_default("FOREST_FILTER_TTL_SECS", config.filter_ttl_secs);
+        let filter_ttl = (filter_ttl_secs > 0).then(|| Duration::from_secs(filter_ttl_secs));
         let filter_store: Option<Arc<dyn FilterStore>> =
             Some(MemFilterStore::new(max_filters) as Arc<dyn FilterStore>);
         let event_filter_manager = Some(EventFilterManager::new());
@@ -187,12 +196,62 @@ impl EthEventHandler {
             filter_store,
             max_filter_results,
             max_filter_height_range,
+            filter_ttl,
             event_filter_manager,
             tipset_filter_manager,
             mempool_filter_manager,
         }
     }
 
+    /// TTL after which idle filters are removed; `None` when expiry is disabled.
+    pub fn filter_ttl(&self) -> Option<Duration> {
+        self.filter_ttl
+    }
+
+    /// Periodically removes filters idle for longer than `ttl`, sweeping every
+    /// [`FILTER_GC_INTERVAL`] or every `ttl`, whichever is shorter.
+    pub async fn run_filter_gc(&self, ttl: Duration) -> ! {
+        let period = ttl.min(FILTER_GC_INTERVAL);
+        let mut interval = tokio::time::interval_at(tokio::time::Instant::now() + period, period);
+        loop {
+            interval.tick().await;
+            self.sweep_expired(ttl);
+        }
+    }
+
+    /// One sweep pass: removes every filter idle for longer than `ttl` from
+    /// the store and from its manager.
+    fn sweep_expired(&self, ttl: Duration) {
+        let Some(filter_store) = &self.filter_store else {
+            return;
+        };
+        let expired = filter_store.remove_expired(ttl);
+        for filter in &expired {
+            if let Err(e) = self.remove_from_manager(filter.as_ref()) {
+                tracing::warn!(
+                    "Failed to remove expired eth filter {:?} from its manager: {e:#}",
+                    filter.id()
+                );
+            }
+        }
+        if !expired.is_empty() {
+            tracing::debug!("Expired {} idle eth filter(s)", expired.len());
+        }
+    }
+
+    /// Completes an install: adds `filter` to the store and returns its id,
+    /// rolling the manager registration back when the store rejects it (e.g.
+    /// at the filter cap).
+    fn register_filter(&self, filter: Arc<dyn Filter>) -> Result<FilterID, Error> {
+        if let Some(filter_store) = &self.filter_store
+            && let Err(err) = filter_store.add(filter.clone())
+        {
+            self.remove_from_manager(filter.as_ref())?;
+            bail!("Adding filter failed: {}", err);
+        }
+        Ok(filter.id().clone())
+    }
+
     // Installs an eth filter based on given filter spec.
     pub fn eth_new_filter(
         &self,
@@ -208,16 +267,7 @@ impl EthEventHandler {
                 .install(pf)
                 .context("Installation error")?;
 
-            if let Some(filter_store) = &self.filter_store
-                && let Err(err) = filter_store.add(filter.clone())
-            {
-                ensure!(
-                    event_filter_manager.remove(filter.id()).is_some(),
-                    "Filter not found"
-                );
-                bail!("Adding filter failed: {}", err);
-            }
-            Ok(filter.id().clone())
+            self.register_filter(filter)
         } else {
             Err(Error::msg("NotSupported"))
         }
@@ -229,13 +279,7 @@ impl EthEventHandler {
     ) -> Result<FilterID, Error> {
         if let Some(manager) = filter_manager {
             let filter = manager.install().context("Installation error")?;
-            if let Some(filter_store) = &self.filter_store
-                && let Err(err) = filter_store.add(filter.clone())
-            {
-                ensure!(manager.remove(filter.id()).is_some(), "Filter not found");
-                bail!("Adding filter failed: {}", err);
-            }
-            Ok(filter.id().clone())
+            self.register_filter(filter)
         } else {
             Err(Error::msg("NotSupported"))
         }
@@ -251,7 +295,7 @@ impl EthEventHandler {
         self.install_filter(self.mempool_filter_manager.as_deref().map(|fm| fm as _))
     }
 
-    fn uninstall_filter(&self, filter: Arc<dyn Filter>) -> Result<(), Error> {
+    fn remove_from_manager(&self, filter: &dyn Filter) -> Result<(), Error> {
         let id = filter.id();
 
         if filter.as_any().is::<EventFilter>() {
@@ -274,23 +318,18 @@ impl EthEventHandler {
                 .context("Failed to remove mempool filter")?;
         }
 
-        self.filter_store
-            .as_ref()
-            .context("Filter store is missing")?
-            .remove(id)
-            .context("Failed to remove filter from store")?;
-
         Ok(())
     }
 
+    /// Uninstalls an eth filter.
     pub fn eth_uninstall_filter(&self, id: &FilterID) -> Result<bool, Error> {
         let store = self
             .filter_store
             .as_ref()
             .context("Filter store is not supported")?;
 
-        if let Ok(filter) = store.get(id) {
-            self.uninstall_filter(filter)?;
+        if let Some(filter) = store.remove(id) {
+            self.remove_from_manager(filter.as_ref())?;
             Ok(true)
         } else {
             Ok(false)
@@ -902,8 +941,8 @@ mod tests {
         assert!(parsed.addresses.is_empty());
     }
 
-    #[test]
-    fn test_eth_new_filter_with_none_address() {
+    #[tokio::test]
+    async fn test_eth_new_filter_with_none_address() {
         let eth_event_handler = EthEventHandler::new();
 
         let filter_spec = EthFilterSpec {
@@ -1209,8 +1248,8 @@ mod tests {
         assert!(result.is_err());
     }
 
-    #[test]
-    fn test_eth_new_filter() {
+    #[tokio::test]
+    async fn test_eth_new_filter() {
         let eth_event_handler = EthEventHandler::new();
 
         let filter_spec = EthFilterSpec {
@@ -1229,8 +1268,8 @@ mod tests {
         assert!(result.is_ok(), "Expected successful filter creation");
     }
 
-    #[test]
-    fn test_eth_new_block_filter() {
+    #[tokio::test]
+    async fn test_eth_new_block_filter() {
         let eth_event_handler = EthEventHandler::new();
         let result = eth_event_handler.eth_new_block_filter();
 
@@ -1269,10 +1308,121 @@ mod tests {
         let pending_tx_filter_id = event_handler.eth_new_pending_transaction_filter().unwrap();
         filter_ids.push(pending_tx_filter_id);
 
-        for filter_id in filter_ids {
-            let result = event_handler.eth_uninstall_filter(&filter_id).unwrap();
+        for filter_id in &filter_ids {
+            let result = event_handler.eth_uninstall_filter(filter_id).unwrap();
             assert!(result, "Uninstalling filter with id {filter_id:?} failed");
         }
+
+        // uninstalling an already-removed filter reports `false` instead of erroring
+        let gone = filter_ids.first().unwrap();
+        assert!(!event_handler.eth_uninstall_filter(gone).unwrap());
+    }
+
+    const TEST_TTL: Duration = Duration::from_hours(1);
+
+    /// Handler whose filters expire after `ttl` without being polled;
+    /// `Duration::ZERO` disables expiry.
+    fn handler_with_ttl(ttl: Duration) -> EthEventHandler {
+        EthEventHandler::from_config(
+            &EventsConfig {
+                filter_ttl_secs: ttl.as_secs(),
+                ..Default::default()
+            },
+            crate::networks::mainnet::ETH_CHAIN_ID,
+            MpoolSubscriber::dummy(),
+        )
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn sweep_expired_removes_idle_filters_of_every_type() {
+        let handler = handler_with_ttl(TEST_TTL);
+        let event_id = handler
+            .eth_new_filter(&EthFilterSpec::default(), 0)
+            .unwrap();
+        let block_id = handler.eth_new_block_filter().unwrap();
+        let pending_id = handler.eth_new_pending_transaction_filter().unwrap();
+
+        // everything has been idle for longer than the TTL
+        tokio::time::advance(TEST_TTL + Duration::from_secs(1)).await;
+        handler.sweep_expired(TEST_TTL);
+
+        // the filters are gone from the store...
+        let store = handler.filter_store.as_ref().unwrap();
+        for id in [&event_id, &block_id, &pending_id] {
+            assert!(store.get(id).is_err(), "filter {id:?} should be expired");
+        }
+        // ...and from their managers
+        let event_manager = handler.event_filter_manager.as_ref().unwrap();
+        assert!(event_manager.remove(&event_id).is_none());
+        let tipset_manager = handler.tipset_filter_manager.as_ref().unwrap();
+        assert!(tipset_manager.remove(&block_id).is_none());
+        let mempool_manager = handler.mempool_filter_manager.as_ref().unwrap();
+        assert!(mempool_manager.remove(&pending_id).is_none());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn sweep_expired_spares_recently_polled_filters() {
+        let handler = handler_with_ttl(TEST_TTL);
+        let polled_id = handler.eth_new_block_filter().unwrap();
+        let idle_id = handler.eth_new_block_filter().unwrap();
+        let store = handler.filter_store.as_ref().unwrap();
+
+        // poll one filter 45 minutes in
+        tokio::time::advance(Duration::from_mins(45)).await;
+        store.get(&polled_id).unwrap();
+
+        // 30 minutes later, the never-polled filter has crossed the TTL
+        // (75 minutes idle) while the polled one has not (30 minutes idle)
+        tokio::time::advance(Duration::from_mins(30)).await;
+        handler.sweep_expired(TEST_TTL);
+
+        assert!(store.get(&idle_id).is_err(), "idle filter should expire");
+        assert!(
+            store.get(&polled_id).is_ok(),
+            "recently polled filter must survive"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn gc_loop_sweeps_idle_filters() {
+        let handler = Arc::new(handler_with_ttl(TEST_TTL));
+        let id = handler.eth_new_block_filter().unwrap();
+        let _gc = tokio::spawn({
+            let handler = handler.clone();
+            async move { handler.run_filter_gc(TEST_TTL).await }
+        });
+        tokio::task::yield_now().await; // let the spawned loop start its interval now
+
+        tokio::time::advance(TEST_TTL + FILTER_GC_INTERVAL).await;
+        tokio::task::yield_now().await; // let the loop process elapsed ticks
+        let store = handler.filter_store.as_ref().unwrap();
+        assert!(store.get(&id).is_err(), "idle filter should be swept");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn short_ttl_sweeps_before_the_default_interval() {
+        let ttl = Duration::from_mins(5);
+        let handler = Arc::new(handler_with_ttl(ttl));
+        let id = handler.eth_new_block_filter().unwrap();
+        let _gc = tokio::spawn({
+            let handler = handler.clone();
+            async move { handler.run_filter_gc(ttl).await }
+        });
+        tokio::task::yield_now().await; // let the spawned loop start its interval now
+
+        // two TTL periods, well inside the 30-minute `FILTER_GC_INTERVAL`
+        tokio::time::advance(ttl * 2 + Duration::from_secs(1)).await;
+        tokio::task::yield_now().await; // let the loop process elapsed ticks
+        let store = handler.filter_store.as_ref().unwrap();
+        assert!(
+            store.get(&id).is_err(),
+            "short-TTL filter must be swept before the default interval elapses"
+        );
+    }
+
+    #[test]
+    fn zero_ttl_disables_filter_expiry() {
+        assert!(handler_with_ttl(Duration::ZERO).filter_ttl().is_none());
     }
 
     #[test]