feat(rpc) Implement Filecoin.EthGetLogs

scripts/tests/api_compare/filter-list

@@ -6,3 +6,5 @@
 !Filecoin.StateReplay
 # TODO(elmattic): https://github.com/ChainSafe/forest/issues/4759
 !Filecoin.EthGetTransactionReceipt
+# TODO(elmattic): https://github.com/ChainSafe/forest/issues/4851
+!Filecoin.EthGetLogs

scripts/tests/api_compare/filter-list-offline

@@ -27,3 +27,5 @@
 !Filecoin.ChainSetHead
 # TODO(elmattic): https://github.com/ChainSafe/forest/issues/4759
 !Filecoin.EthGetTransactionReceipt
+# TODO(elmattic): https://github.com/ChainSafe/forest/issues/4851
+!Filecoin.EthGetLogs

src/chain/store/chain_store.rs

@@ -5,8 +5,7 @@ use std::sync::Arc;
 
 use crate::blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta};
 use crate::fil_cns;
-use crate::interpreter::BlockMessages;
-use crate::interpreter::VMTrace;
+use crate::interpreter::{BlockMessages, EventCache, VMTrace};
 use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage};
 use crate::networks::{ChainConfig, Height};
@@ -16,6 +15,7 @@ use crate::shim::{
     address::Address, econ::TokenAmount, executor::Receipt, message::Message,
     state_tree::StateTree, version::NetworkVersion,
 };
+use crate::state_manager::StateOutput;
 use crate::utils::db::{BlockstoreExt, CborStoreExt};
 use ahash::{HashMap, HashMapExt, HashSet};
 use anyhow::Context;
@@ -340,7 +340,7 @@ where
             // state-root without caching.
             let genesis_timestamp = heaviest_tipset.genesis(&chain_index.db)?.timestamp;
             let beacon = Arc::new(chain_config.get_beacon_schedule(genesis_timestamp));
-            let (state, _) = crate::state_manager::apply_block_messages(
+            let StateOutput { state_root, .. } = crate::state_manager::apply_block_messages(
                 genesis_timestamp,
                 Arc::clone(&chain_index),
                 Arc::clone(&chain_config),
@@ -353,9 +353,10 @@ where
                 Arc::clone(&heaviest_tipset),
                 crate::state_manager::NO_CALLBACK,
                 VMTrace::NotTraced,
+                EventCache::NotCached,
             )
             .map_err(|e| Error::Other(e.to_string()))?;
-            return Ok((heaviest_tipset, state));
+            return Ok((heaviest_tipset, state_root));
         }
 
         let next_ts = chain_index

src/cli_shared/cli/client.rs

@@ -66,6 +66,7 @@ pub struct Client {
     pub load_actors: bool,
     /// `TTL` to set for Ethereum `Hash` to `Cid` entries or `None` to never reclaim them.
     pub eth_mapping_ttl: Option<u32>,
+    pub enable_event_caching: bool,
 }
 
 impl Default for Client {
@@ -93,6 +94,7 @@ impl Default for Client {
             ),
             load_actors: true,
             eth_mapping_ttl: None,
+            enable_event_caching: true,
         }
     }
 }

src/cli_shared/cli/config.rs

@@ -52,6 +52,10 @@ impl Config {
     pub fn db_config(&self) -> &DbConfig {
         &self.parity_db
     }
+
+    pub fn chain(&self) -> &NetworkChain {
+        &self.chain
+    }
 }
 
 #[cfg(test)]

src/cli_shared/cli/mod.rs

@@ -154,6 +154,9 @@ pub struct CliOpts {
     /// Skip loading actors from the actors bundle.
     #[arg(long)]
     pub skip_load_actors: bool,
+    /// Cache events in the tipset state cache (default: true)
+    #[arg(long)]
+    pub enable_event_caching: Option<bool>,
 }
 
 impl CliOpts {
@@ -224,6 +227,9 @@ impl CliOpts {
         }
 
         cfg.client.load_actors = !self.skip_load_actors;
+        cfg.client.enable_event_caching = self
+            .enable_event_caching
+            .unwrap_or(cfg.client.enable_event_caching);
 
         Ok((cfg, path))
     }

src/cli_shared/mod.rs

@@ -18,7 +18,7 @@ pub use tikv_jemallocator;
 
 /// Gets chain data directory
 pub fn chain_path(config: &Config) -> PathBuf {
-    PathBuf::from(&config.client.data_dir).join(config.chain.to_string())
+    PathBuf::from(&config.client.data_dir).join(config.chain().to_string())
 }
 
 /// Gets car db path
@@ -57,23 +57,23 @@ mod tests {
         let (config_path, config) = read_config(None, None).unwrap();
 
         assert!(config_path.is_none());
-        assert_eq!(config.chain, NetworkChain::Mainnet);
+        assert_eq!(config.chain(), &NetworkChain::Mainnet);
     }
 
     #[test]
     fn read_config_calibnet_override() {
         let (config_path, config) = read_config(None, Some(NetworkChain::Calibnet)).unwrap();
 
         assert!(config_path.is_none());
-        assert_eq!(config.chain, NetworkChain::Calibnet);
+        assert_eq!(config.chain(), &NetworkChain::Calibnet);
     }
 
     #[test]
     fn read_config_butterflynet_override() {
         let (config_path, config) = read_config(None, Some(NetworkChain::Butterflynet)).unwrap();
 
         assert!(config_path.is_none());
-        assert_eq!(config.chain, NetworkChain::Butterflynet);
+        assert_eq!(config.chain(), &NetworkChain::Butterflynet);
     }
 
     #[test]
@@ -86,7 +86,7 @@ mod tests {
         let (config_path, config) = read_config(Some(&path), None).unwrap();
 
         assert_eq!(config_path.unwrap(), ConfigPath::Cli(path));
-        assert_eq!(config.chain, NetworkChain::Mainnet);
+        assert_eq!(config.chain(), &NetworkChain::Mainnet);
         assert_eq!(config, default_config);
     }
 }

src/daemon/main.rs

@@ -101,7 +101,7 @@ where
         }
         check_for_unknown_keys(path.to_path_buf(), &cfg);
     } else {
-        info!("Using default {} config", cfg.chain);
+        info!("Using default {} config", cfg.chain());
     }
     if opts.dry_run {
         return Ok(());

src/daemon/mod.rs

@@ -147,7 +147,7 @@ pub(super) async fn start(
     config: Config,
     shutdown_send: mpsc::Sender<()>,
 ) -> anyhow::Result<()> {
-    let chain_config = Arc::new(ChainConfig::from_chain(&config.chain));
+    let chain_config = Arc::new(ChainConfig::from_chain(config.chain()));
     if chain_config.is_testnet() {
         CurrentNetwork::set_global(Network::Testnet);
     }
@@ -189,7 +189,7 @@ pub(super) async fn start(
     load_all_forest_cars(&db, &forest_car_db_dir)?;
 
     if config.client.load_actors && !opts.stateless {
-        load_actor_bundles(&db, &config.chain).await?;
+        load_actor_bundles(&db, config.chain()).await?;
     }
 
     let mut services = JoinSet::new();
@@ -289,14 +289,15 @@ pub(super) async fn start(
         Arc::clone(&chain_store),
         Arc::clone(&chain_config),
         Arc::new(config.sync.clone()),
+        config.client.enable_event_caching,
     )?;
 
     let state_manager = Arc::new(sm);
 
     let network_name = get_network_name_from_genesis(&genesis_header, &state_manager)?;
 
     info!("Using network :: {}", get_actual_chain_name(&network_name));
-    utils::misc::display_chain_logo(&config.chain);
+    utils::misc::display_chain_logo(config.chain());
     let (tipset_sender, tipset_receiver) = flume::bounded(20);
 
     // if bootstrap peers are not set, set them
@@ -424,7 +425,10 @@ pub(super) async fn start(
                 ))
                 .expect("F3 lease manager should not have been initialized before");
             let chain_config = chain_config.clone();
-            let default_f3_root = config.client.data_dir.join(format!("f3/{}", config.chain));
+            let default_f3_root = config
+                .client
+                .data_dir
+                .join(format!("f3/{}", config.chain()));
             let crate::f3::F3Options {
                 chain_finality,
                 bootstrap_epoch,
@@ -553,7 +557,7 @@ async fn set_snapshot_path_if_needed(
     }
 
     let vendor = snapshot::TrustedVendor::default();
-    let chain = &config.chain;
+    let chain = config.chain();
 
     // What height is our chain at right now, and what network version does that correspond to?
     let network_version = chain_config.network_version(epoch);

src/db/migration/v0_19_0.rs

@@ -154,7 +154,7 @@ async fn create_state_manager_and_populate(config: Config, db_name: String) -> a
     let forest_car_db_dir = db_root_dir.join(CAR_DB_DIR_NAME);
     load_all_forest_cars(&db, &forest_car_db_dir)?;
 
-    let chain_config = Arc::new(ChainConfig::from_chain(&config.chain));
+    let chain_config = Arc::new(ChainConfig::from_chain(config.chain()));
 
     let genesis_header = read_genesis_header(
         config.client.genesis_file.as_deref(),
@@ -173,8 +173,9 @@ async fn create_state_manager_and_populate(config: Config, db_name: String) -> a
 
     let state_manager = StateManager::new(
         Arc::clone(&chain_store),
-        Arc::clone(&chain_config),
+        chain_config,
         Arc::new(config.sync.clone()),
+        false,
     )?;
 
     populate_eth_mappings(&state_manager, &chain_store.heaviest_tipset())?;

src/interpreter/vm.rs

@@ -17,7 +17,7 @@ use crate::networks::{ChainConfig, NetworkChain};
 use crate::shim::{
     address::Address,
     econ::TokenAmount,
-    executor::{ApplyRet, Receipt},
+    executor::{ApplyRet, Receipt, StampedEvent},
     externs::{Rand, RandWrapper},
     machine::MultiEngine,
     message::{Message, Message_v3},
@@ -76,6 +76,8 @@ type ForestExecutorV4<DB> = DefaultExecutor_v4<ForestKernelV4<DB>>;
 
 pub type ApplyResult = anyhow::Result<(ApplyRet, Duration)>;
 
+pub type ApplyBlockResult = anyhow::Result<(Vec<Receipt>, Vec<Vec<StampedEvent>>), anyhow::Error>;
+
 /// Comes from <https://github.com/filecoin-project/lotus/blob/v1.23.2/chain/vm/fvm.go#L473>
 pub const IMPLICIT_MESSAGE_GAS_LIMIT: i64 = i64::MAX / 2;
 
@@ -351,8 +353,10 @@ where
         messages: &[BlockMessages],
         epoch: ChainEpoch,
         mut callback: Option<impl FnMut(MessageCallbackCtx<'_>) -> anyhow::Result<()>>,
-    ) -> Result<Vec<Receipt>, anyhow::Error> {
+        enable_event_caching: EventCache,
+    ) -> ApplyBlockResult {
         let mut receipts = Vec::new();
+        let mut events = Vec::new();
         let mut processed = HashSet::<Cid>::default();
 
         for block in messages.iter() {
@@ -383,6 +387,10 @@ where
                 let msg_receipt = ret.msg_receipt();
                 receipts.push(msg_receipt.clone());
 
+                if enable_event_caching.is_cached() {
+                    events.push(ret.events());
+                }
+
                 // Add processed Cid to set of processed messages
                 processed.insert(cid);
                 Ok(())
@@ -428,7 +436,7 @@ where
             tracing::error!("End of epoch cron failed to run: {}", e);
         }
 
-        Ok(receipts)
+        Ok((receipts, events))
     }
 
     /// Applies single message through VM and returns result from execution.
@@ -590,3 +598,21 @@ impl VMTrace {
         matches!(self, VMTrace::Traced)
     }
 }
+
+/// Caching events will use more memory.
+/// This controls whether we should cache or not events when a state transition is performed.
+#[derive(Default, Clone, Copy)]
+pub enum EventCache {
+    /// Store event in the tipset state cache
+    Cached,
+    /// Do not cache event
+    #[default]
+    NotCached,
+}
+
+impl EventCache {
+    /// Should event be kept in cache?
+    pub fn is_cached(&self) -> bool {
+        matches!(self, EventCache::Cached)
+    }
+}