fix: optimize collect_events in Eth RPC

src/chain/store/chain_store.rs

@@ -275,6 +275,30 @@ where
         }
     }
 
+    pub fn load_child_tipset(&self, ts: &Tipset) -> Result<Tipset, Error> {
+        let head = self.heaviest_tipset();
+        if head.parents() == ts.key() {
+            Ok(head)
+        } else if head.epoch() > ts.epoch() {
+            let maybe_child = self.chain_index().tipset_by_height(
+                ts.epoch() + 1,
+                head,
+                ResolveNullTipset::TakeNewer,
+            )?;
+            if maybe_child.parents() == ts.key() {
+                Ok(maybe_child)
+            } else {
+                Err(Error::NotFound(
+                    format!("child of tipset@{}", ts.epoch()).into(),
+                ))
+            }
+        } else {
+            Err(Error::NotFound(
+                format!("child of tipset@{}", ts.epoch()).into(),
+            ))
+        }
+    }
+
     /// Determines if provided tipset is heavier than existing known heaviest
     /// tipset
     fn update_heaviest(&self, ts: Tipset) -> Result<(), Error> {

src/chain/store/errors.rs

@@ -1,7 +1,7 @@
 // Copyright 2019-2026 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use std::fmt::Debug;
+use std::{borrow::Cow, fmt::Debug};
 
 use crate::blocks::CreateTipsetError;
 use cid::Error as CidErr;
@@ -17,7 +17,7 @@ pub enum Error {
     UndefinedKey(String),
     /// Key not found in database
     #[error("{0} not found")]
-    NotFound(String),
+    NotFound(Cow<'static, str>),
     /// Error originating constructing blockchain structures
     #[error(transparent)]
     Blockchain(#[from] CreateTipsetError),
@@ -29,7 +29,7 @@ pub enum Error {
     Cid(#[from] CidErr),
     /// Amt error
     #[error("State error: {0}")]
-    State(String),
+    State(Cow<'static, str>),
     /// Other chain error
     #[error("{0}")]
     Other(String),
@@ -43,7 +43,7 @@ impl From<EncErr> for Error {
 
 impl From<AmtErr> for Error {
     fn from(e: AmtErr) -> Error {
-        Error::State(e.to_string())
+        Error::state(e.to_string())
     }
 }
 
@@ -72,7 +72,7 @@ impl<T> From<flume::SendError<T>> for Error {
 }
 
 impl Error {
-    pub fn state(msg: impl std::fmt::Display) -> Self {
-        Self::State(msg.to_string())
+    pub fn state(msg: impl Into<Cow<'static, str>>) -> Self {
+        Self::State(msg.into())
     }
 }

src/dev/subcommands/state_cmd.rs

@@ -90,11 +90,7 @@ impl ComputeCommand {
                 chain_store.heaviest_tipset(),
                 ResolveNullTipset::TakeOlder,
             )?;
-            let ts_next = chain_store.chain_index().tipset_by_height(
-                epoch + 1,
-                chain_store.heaviest_tipset(),
-                ResolveNullTipset::TakeNewer,
-            )?;
+            let ts_next = chain_store.load_child_tipset(&ts)?;
             db.resume_tracking();
             SettingsStoreExt::write_obj(
                 &db.tracker,
@@ -113,7 +109,6 @@ impl ComputeCommand {
         let StateOutput {
             state_root,
             receipt_root,
-            ..
         } = state_manager
             .compute_tipset_state(ts, crate::state_manager::NO_CALLBACK, VMTrace::NotTraced)
             .await?;

src/rpc/methods/eth.rs

@@ -52,7 +52,7 @@ use crate::shim::gas::GasOutputs;
 use crate::shim::message::Message;
 use crate::shim::trace::{CallReturn, ExecutionEvent};
 use crate::shim::{clock::ChainEpoch, state_tree::StateTree};
-use crate::state_manager::{StateLookupPolicy, VMFlush};
+use crate::state_manager::{ExecutedMessage, ExecutedTipset, StateLookupPolicy, VMFlush};
 use crate::utils::cache::SizeTrackingLruCache;
 use crate::utils::db::BlockstoreExt as _;
 use crate::utils::encoding::from_slice_with_fallback;
@@ -497,16 +497,28 @@ impl Block {
             let block_number = EthInt64(tipset.epoch());
             let block_hash: EthHash = block_cid.into();
 
-            let (state_root, msgs_and_receipts) = execute_tipset(&ctx, &tipset).await?;
-
+            let ExecutedTipset {
+                state_root,
+                executed_messages,
+            } = ctx
+                .state_manager
+                .load_executed_tipset_without_events(&tipset)
+                .await?;
+            let has_transactions = !executed_messages.is_empty();
             let state_tree = ctx.state_manager.get_state_tree(&state_root)?;
 
             let mut full_transactions = vec![];
             let mut gas_used = 0;
-            for (i, (msg, receipt)) in msgs_and_receipts.iter().enumerate() {
+            for (
+                i,
+                ExecutedMessage {
+                    message, receipt, ..
+                },
+            ) in executed_messages.into_iter().enumerate()
+            {
                 let ti = EthUint64(i as u64);
                 gas_used += receipt.gas_used();
-                let smsg = match msg {
+                let smsg = match message {
                     ChainMessage::Signed(msg) => msg.clone(),
                     ChainMessage::Unsigned(msg) => {
                         let sig = Signature::new_bls(vec![]);
@@ -538,7 +550,7 @@ impl Block {
                     .into(),
                 gas_used: EthUint64(gas_used),
                 transactions: Transactions::Full(full_transactions),
-                ..Block::new(!msgs_and_receipts.is_empty(), tipset.len())
+                ..Block::new(has_transactions, tipset.len())
             };
             ETH_BLOCK_CACHE.push(block_cid.into(), b.clone());
             b
@@ -959,28 +971,6 @@ fn resolve_block_hash_tipset<DB: Blockstore>(
     Ok(ts)
 }
 
-async fn execute_tipset<DB: Blockstore + Send + Sync + 'static>(
-    data: &Ctx<DB>,
-    tipset: &Tipset,
-) -> Result<(Cid, Vec<(ChainMessage, Receipt)>)> {
-    let msgs = data.chain_store().messages_for_tipset(tipset)?;
-
-    let (state_root, _) = data
-        .state_manager
-        .tipset_state(tipset, StateLookupPolicy::Enabled)
-        .await?;
-    let receipts = data.state_manager.tipset_message_receipts(tipset).await?;
-
-    if msgs.len() != receipts.len() {
-        bail!("receipts and message array lengths didn't match for tipset: {tipset:?}")
-    }
-
-    Ok((
-        state_root,
-        msgs.into_iter().zip(receipts.into_iter()).collect(),
-    ))
-}
-
 pub fn is_eth_address(addr: &VmAddress) -> bool {
     if addr.protocol() != Protocol::Delegated {
         return false;
@@ -1426,19 +1416,31 @@ async fn get_block_receipts<DB: Blockstore + Send + Sync + 'static>(
     let ts_key = ts_ref.key();
 
     // Execute the tipset to get the messages and receipts
-    let (state_root, msgs_and_receipts) = execute_tipset(ctx, &ts_ref).await?;
+    let ExecutedTipset {
+        state_root,
+        executed_messages,
+    } = ctx
+        .state_manager
+        .load_executed_tipset_without_events(&ts_ref)
+        .await?;
 
     // Load the state tree
     let state_tree = ctx.state_manager.get_state_tree(&state_root)?;
 
-    let mut eth_receipts = Vec::with_capacity(msgs_and_receipts.len());
-    for (i, (msg, receipt)) in msgs_and_receipts.into_iter().enumerate() {
+    let mut eth_receipts = Vec::with_capacity(executed_messages.len());
+    for (
+        i,
+        ExecutedMessage {
+            message, receipt, ..
+        },
+    ) in executed_messages.into_iter().enumerate()
+    {
         let tx = new_eth_tx(
             ctx,
             &state_tree,
             ts_ref.epoch(),
             &ts_key.cid()?,
-            &msg.cid(),
+            &message.cid(),
             i as u64,
         )?;
 
@@ -1929,9 +1931,17 @@ async fn eth_fee_history<B: Blockstore + Send + Sync + 'static>(
         .take(block_count as _)
     {
         let base_fee = &ts.block_headers().first().parent_base_fee;
-        let (_state_root, messages_and_receipts) = execute_tipset(&ctx, &ts).await?;
-        let mut tx_gas_rewards = Vec::with_capacity(messages_and_receipts.len());
-        for (message, receipt) in messages_and_receipts {
+        let ExecutedTipset {
+            executed_messages, ..
+        } = ctx
+            .state_manager
+            .load_executed_tipset_without_events(&ts)
+            .await?;
+        let mut tx_gas_rewards = Vec::with_capacity(executed_messages.len());
+        for ExecutedMessage {
+            message, receipt, ..
+        } in executed_messages
+        {
             let premium = message.effective_gas_premium(base_fee);
             tx_gas_rewards.push(GasReward {
                 gas_used: receipt.gas_used(),