Fix ChainGetEvents RPC method

Author: elmattic

src/daemon/db_util.rs

@@ -326,9 +326,11 @@ where
             .compute_tipset_state(Arc::new(ts), NO_CALLBACK, VMTrace::NotTraced)
             .await?;
         for events_root in state_output.events_roots.iter() {
-            println!("Indexing events root @{}: {}", epoch, events_root);
+            if let Some(cid) = events_root {
+                println!("Indexing events root @{}: {}", epoch, cid);
 
-            state_manager.chain_store().put_index(events_root, &tsk)?;
+                state_manager.chain_store().put_index(cid, &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
@@ -530,9 +530,11 @@ where
                     .compute_tipset_state(Arc::clone(tipset), NO_CALLBACK, VMTrace::NotTraced)
                     .await?;
                 for events_root in state_output.events_roots.iter() {
-                    trace!("Indexing events root @{}: {}", tipset.epoch(), events_root);
+                    if let Some(cid) = events_root {
+                        trace!("Indexing events root @{}: {}", tipset.epoch(), cid);
 
-                    self.chain_store().put_index(events_root, key)?;
+                        self.chain_store().put_index(cid, key)?;
+                    }
                 }
                 let ts_state = state_output.into();
                 trace!("Completed tipset state calculation {:?}", tipset.cids());

src/tool/subcommands/index_cmd.rs

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