store the fvm events in blockstore and fetch from blockstore if not available in cache [skip ci]

Author: akaladarshi

src/chain/store/chain_store.rs

@@ -203,7 +203,7 @@ where
         Ok(())
     }
 
-    pub fn get_tipset_key(&self, key: &Cid) -> Result<Option<TipsetKey>, Error> {
+    pub fn get_tipset_key_by_events_root(&self, key: &Cid) -> Result<Option<TipsetKey>, Error> {
         Ok(self.indices.read_obj(key)?)
     }
 

src/rpc/methods/chain.rs

@@ -239,8 +239,8 @@ impl RpcMethod<1> for ChainGetEvents {
         let tsk = ctx
             .state_manager
             .chain_store()
-            .get_tipset_key(&root_cid)?
-            .with_context(|| format!("can't find events with cid {root_cid}"))?;
+            .get_tipset_key_by_events_root(&root_cid)?
+            .with_context(|| format!("can't find tipset for events root {root_cid}"))?;
 
         let ts = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?;
 

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

@@ -45,7 +45,6 @@ use anyhow::{Context, Error, anyhow, bail, ensure};
 use cid::Cid;
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_encoding::IPLD_RAW;
-use itertools::Itertools;
 use serde::*;
 use std::ops::RangeInclusive;
 use std::sync::Arc;
@@ -384,11 +383,13 @@ impl EthEventHandler {
             .filter(|(cid, _)| cid.as_ref() == Some(events_root))
             .map(|(_, v)| v);
 
-        let chain_events = filtered_events
+        // Do NOT deduplicate events - the AMT can legitimately contain duplicate events
+        // if a contract emits the same event multiple times. We must preserve the exact
+        // order and count of events as stored in the AMT.
+        let chain_events: Vec<Event> = filtered_events
             .into_iter()
             .flat_map(|events| events.into_iter())
             .map(Into::into)
-            .unique()
             .collect();
 
         Ok(chain_events)

src/shim/executor.rs

@@ -7,7 +7,7 @@ use crate::shim::{
 };
 use crate::utils::get_size::{GetSize, vec_heap_size_with_fn_helper};
 use cid::Cid;
-use fil_actors_shared::fvm_ipld_amt::Amtv0;
+use fil_actors_shared::fvm_ipld_amt::{Amt, Amtv0};
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_encoding::RawBytes;
 use fvm_shared2::receipt::Receipt as Receipt_v2;
@@ -339,6 +339,7 @@ impl ActorEvent {
 
 /// Event with extra information stamped by the FVM.
 #[derive(Clone, Debug, Serialize)]
+#[serde(untagged)]
 pub enum StampedEvent {
     V3(StampedEvent_v3),
     V4(StampedEvent_v4),
@@ -385,6 +386,32 @@ impl StampedEvent {
             Self::V4(v4) => v4.event.clone().into(),
         }
     }
+
+    /// Loads events directly from the events AMT root CID.
+    /// Returns events in the exact order they are stored in the AMT.
+    pub fn get_events<DB: Blockstore>(
+        db: &DB,
+        events_root: &Cid,
+    ) -> anyhow::Result<Vec<StampedEvent>> {
+        let mut events = Vec::new();
+
+        // Try StampedEvent_v4 first (StampedEvent_v4 and StampedEvent_v3 are identical, use v4 here)
+        if let Ok(amt) = Amt::<StampedEvent_v4, _>::load(events_root, db) {
+            amt.for_each(|_, event| {
+                events.push(StampedEvent::V4(event.clone()));
+                Ok(())
+            })?;
+        } else {
+            // Fallback to StampedEvent_v3
+            let amt = Amt::<StampedEvent_v3, _>::load(events_root, db)?;
+            amt.for_each(|_, event| {
+                events.push(StampedEvent::V3(event.clone()));
+                Ok(())
+            })?;
+        }
+
+        Ok(events)
+    }
 }
 
 #[cfg(test)]

src/state_manager/mod.rs

@@ -61,14 +61,14 @@ use crate::utils::get_size::{
     GetSize, vec_heap_size_helper, vec_with_stack_only_item_heap_size_helper,
 };
 use ahash::{HashMap, HashMapExt};
-use anyhow::{Context as _, bail};
+use anyhow::{Context as _, bail, ensure};
 use bls_signatures::{PublicKey as BlsPublicKey, Serialize as _};
 use chain_rand::ChainRand;
 use cid::Cid;
 pub use circulating_supply::GenesisInfo;
 use fil_actor_verifreg_state::v12::DataCap;
 use fil_actor_verifreg_state::v13::ClaimID;
-use fil_actors_shared::fvm_ipld_amt::Amtv0 as Amt;
+use fil_actors_shared::fvm_ipld_amt::{Amt, Amtv0};
 use fil_actors_shared::fvm_ipld_bitfield::BitField;
 use fil_actors_shared::v12::runtime::DomainSeparationTag;
 use fil_actors_shared::v13::runtime::Policy;
@@ -91,6 +91,7 @@ use tracing::{error, info, instrument, trace, warn};
 pub use utils::is_valid_for_sending;
 
 const DEFAULT_TIPSET_CACHE_SIZE: NonZeroUsize = nonzero!(1024usize);
+const EVENTS_AMT_BITWIDTH: u32 = 5;
 
 /// Intermediary for retrieving state objects and updating actor states.
 type CidPair = (Cid, Cid);
@@ -560,11 +561,24 @@ where
         let ts = tipset.clone();
         let this = Arc::clone(self);
         let cids = tipset.cids();
+        let events_root = events_root.cloned();
         self.receipt_event_cache_handler
             .get_events_or_else(
                 key,
                 Box::new(move || {
                     Box::pin(async move {
+                        // If the events are not in the cache, try to load them from the blockstore
+                        if let Some(events_root) = events_root
+                            && let Ok(stamped_events) =
+                                StampedEvent::get_events(this.blockstore(), &events_root)
+                        {
+                            return Ok(StateEvents {
+                                events: vec![stamped_events],
+                                roots: vec![Some(events_root)],
+                            });
+                        }
+
+                        // If the events are neither in the cache nor in the blockstore, compute them.
                         let state_out = this
                             .compute_tipset_state(ts, NO_CALLBACK, VMTrace::NotTraced)
                             .await?;
@@ -1986,8 +2000,28 @@ where
         let (receipts, events, events_roots) =
             vm.apply_block_messages(&block_messages, epoch, callback)?;
 
-        // step 5: construct receipt root from receipts and flush the state-tree
-        let receipt_root = Amt::new_from_iter(chain_index.db(), receipts)?;
+        // step 5: construct receipt root from receipts
+        let receipt_root = Amtv0::new_from_iter(chain_index.db(), receipts)?;
+
+        // step 6: store events AMTs in the blockstore
+        for (msg_events, events_root) in events.iter().zip(events_roots.iter()) {
+            if let Some(event_root) = events_root {
+                // Store the events AMT - the root CID should match the one computed by FVM
+                let derived_event_root = Amt::new_from_iter_with_bit_width(
+                    chain_index.db(),
+                    EVENTS_AMT_BITWIDTH,
+                    msg_events.iter(),
+                )
+                .map_err(|e| Error::Other(format!("failed to store events AMT: {e}")))?;
+
+                // Verify the stored root matches the FVM-computed root
+                ensure!(
+                    derived_event_root.eq(event_root),
+                    "Events AMT root mismatch: derived={derived_event_root}, actual={event_root}."
+                );
+            }
+        }
+
         let state_root = vm.flush()?;
 
         Ok(StateOutput {