Fix Filecoin.ChainGetEvents method (#5593)

Author: elmattic

scripts/tests/api_compare/filter-list

@@ -2,5 +2,3 @@
 # They should be considered bugged, and not used until the root cause is resolved.
 # Disable until next Lotus release with go-f3 0.8.0
 !Filecoin.F3GetManifest
-# CustomCheckFailed in Forest: https://github.com/ChainSafe/forest/issues/5579
-!Filecoin.ChainGetEvents

scripts/tests/api_compare/filter-list-offline

@@ -27,5 +27,3 @@
 !Filecoin.EthGetBlockByNumber
 !eth_getBlockByNumber
 !Filecoin.ChainSetHead
-# CustomCheckFailed in Forest: https://github.com/ChainSafe/forest/issues/5579
-!Filecoin.ChainGetEvents

src/daemon/db_util.rs

@@ -325,7 +325,7 @@ where
         let state_output = state_manager
             .compute_tipset_state(Arc::new(ts), NO_CALLBACK, VMTrace::NotTraced)
             .await?;
-        for events_root in state_output.events_roots.iter() {
+        for events_root in state_output.events_roots.iter().flatten() {
             println!("Indexing events root @{}: {}", epoch, events_root);
 
             state_manager.chain_store().put_index(events_root, &tsk)?;

src/interpreter/vm.rs

@@ -77,7 +77,7 @@ type ForestExecutorV4<DB> = DefaultExecutor_v4<ForestKernelV4<DB>>;
 pub type ApplyResult = anyhow::Result<(ApplyRet, Duration)>;
 
 pub type ApplyBlockResult =
-    anyhow::Result<(Vec<Receipt>, Vec<Vec<StampedEvent>>, Vec<Cid>), anyhow::Error>;
+    anyhow::Result<(Vec<Receipt>, Vec<Vec<StampedEvent>>, Vec<Option<Cid>>), 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;
@@ -357,7 +357,7 @@ where
     ) -> ApplyBlockResult {
         let mut receipts = Vec::new();
         let mut events = Vec::new();
-        let mut events_roots = Vec::new();
+        let mut events_roots: Vec<Option<Cid>> = Vec::new();
         let mut processed = HashSet::default();
 
         for block in messages.iter() {
@@ -388,12 +388,8 @@ where
                 let msg_receipt = ret.msg_receipt();
                 receipts.push(msg_receipt.clone());
 
-                if let Some(events_root) = ret.msg_receipt().events_root() {
-                    events_roots.push(events_root);
-                    events.push(ret.events());
-                } else {
-                    events.push(ret.events());
-                }
+                events_roots.push(ret.msg_receipt().events_root());
+                events.push(ret.events());
 
                 // Add processed Cid to set of processed messages
                 processed.insert(cid);

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

@@ -374,11 +374,13 @@ impl EthEventHandler {
             .tipset_state_events(tipset, Some(events_root))
             .await?;
 
+        ensure!(state_events.roots.len() == state_events.events.len());
+
         let filtered_events = state_events
             .roots
             .into_iter()
             .zip(state_events.events)
-            .filter(|(cid, _)| cid == events_root)
+            .filter(|(cid, _)| cid.as_ref() == Some(events_root))
             .map(|(_, v)| v);
 
         let mut chain_events = vec![];

src/state_manager/mod.rs

@@ -92,7 +92,7 @@ pub struct StateOutput {
     pub state_root: Cid,
     pub receipt_root: Cid,
     pub events: Vec<Vec<StampedEvent>>,
-    pub events_roots: Vec<Cid>,
+    pub events_roots: Vec<Option<Cid>>,
 }
 
 #[derive(Clone)]
@@ -124,7 +124,7 @@ impl From<StateOutput> for StateOutputValue {
 #[derive(Clone)]
 pub struct StateEvents {
     pub events: Vec<Vec<StampedEvent>>,
-    pub roots: Vec<Cid>,
+    pub roots: Vec<Option<Cid>>,
 }
 
 // Various structures for implementing the tipset state cache
@@ -529,7 +529,7 @@ where
                 let state_output = self
                     .compute_tipset_state(Arc::clone(tipset), NO_CALLBACK, VMTrace::NotTraced)
                     .await?;
-                for events_root in state_output.events_roots.iter() {
+                for events_root in state_output.events_roots.iter().flatten() {
                     trace!("Indexing events root @{}: {}", tipset.epoch(), events_root);
 
                     self.chain_store().put_index(events_root, key)?;

src/tool/subcommands/index_cmd.rs

@@ -96,7 +96,7 @@ impl IndexCommands {
                     let state_output = state_manager
                         .compute_tipset_state(Arc::new(ts), NO_CALLBACK, VMTrace::NotTraced)
                         .await?;
-                    for events_root in state_output.events_roots.iter() {
+                    for events_root in state_output.events_roots.iter().flatten() {
                         println!("Indexing events root @{}: {}", epoch, events_root);
 
                         chain_store.put_index(events_root, &tsk)?;