forest/state_manager/

execution.rs

// Copyright 2019-2026 ChainSafe Systems
// SPDX-License-Identifier: Apache-2.0, MIT

use super::state_computation::{
    TipsetExecutor, apply_block_messages_blocking, validate_tipsets_blocking,
};
use super::utils::structured;
use super::*;
use crate::interpreter::{CalledAt, VMTrace};
use crate::rpc::eth::types::CallSource;
use crate::rpc::state::{ApiInvocResult, MessageGasCost};
use anyhow::{Context as _, bail};
use num_traits::identities::Zero;
use std::ops::RangeInclusive;

impl StateManager {
    /// Replays the given message and returns the result of executing the
    /// indicated message, assuming it was executed in the indicated tipset.
    ///
    /// Served from the shared tipset trace cache, so concurrent replays of
    /// messages in the same tipset share a single traced execution. Unlike
    /// Lotus, which halts at the target message, this executes the whole
    /// tipset — the coalescing depends on it, don't port the halt back.
    /// Consequently, failures after the target message also fail the replay.
    pub async fn replay(
        &self,
        ts: Tipset,
        mcid: Cid,
        source: CallSource,
    ) -> Result<ApiInvocResult, Error> {
        let (_, trace) = self
            .execution_trace(&ts, source)
            .await
            .map_err(|e| Error::Other(format!("unexpected error during execution : {e}")))?;
        trace
            .iter()
            .find(|r| r.msg_cid == mcid)
            .map(|r| (**r).clone())
            .ok_or_else(|| Error::Other("failed to replay".into()))
    }

    /// Replays a tipset up to a target message, capturing the state root before
    /// and after execution.
    pub async fn replay_for_prestate(
        &self,
        ts: Tipset,
        target_message_cid: Cid,
    ) -> Result<(Cid, ApiInvocResult, Cid), Error> {
        let permit = self.replay_permit().await;
        let this = self.shallow_clone();
        tokio::task::spawn_blocking(move || {
            let _permit = permit;
            this.replay_for_prestate_blocking(ts, target_message_cid)
        })
        .await
        .map_err(|e| Error::Other(format!("{e}")))?
    }

    fn replay_for_prestate_blocking(
        &self,
        ts: Tipset,
        target_msg_cid: Cid,
    ) -> Result<(Cid, ApiInvocResult, Cid), Error> {
        if ts.epoch() == 0 {
            return Err(Error::Other(
                "cannot trace messages in the genesis block".into(),
            ));
        }

        let genesis_timestamp = self.chain_store().genesis_block_header().timestamp;
        let exec = TipsetExecutor::new(
            self.chain_index().shallow_clone(),
            self.chain_config().shallow_clone(),
            self.beacon_schedule().shallow_clone(),
            &self.engine,
            ts.shallow_clone(),
        );
        let mut no_cb = NO_CALLBACK;
        let (parent_state, epoch, block_messages) =
            exec.prepare_parent_state_blocking(genesis_timestamp, VMTrace::NotTraced, &mut no_cb)?;

        Ok(stacker::grow(64 << 20, || {
            let mut vm =
                exec.create_vm(parent_state, epoch, ts.min_timestamp(), VMTrace::NotTraced)?;
            let mut processed = ahash::HashSet::default();

            for block in block_messages.iter() {
                let mut penalty = TokenAmount::zero();
                let mut gas_reward = TokenAmount::zero();

                for msg in block.messages.iter() {
                    let cid = msg.cid();
                    if processed.contains(&cid) {
                        continue;
                    }

                    processed.insert(cid);

                    if cid == target_msg_cid {
                        let pre_root = vm.flush()?;
                        let mut traced_vm =
                            exec.create_vm(pre_root, epoch, ts.min_timestamp(), VMTrace::Traced)?;
                        let (ret, duration) = traced_vm.apply_message(msg)?;
                        let post_root = traced_vm.flush()?;

                        return Ok((
                            pre_root,
                            ApiInvocResult {
                                msg_cid: cid,
                                msg: msg.message().clone(),
                                msg_rct: Some(ret.msg_receipt()),
                                error: ret.failure_info().unwrap_or_default(),
                                duration: duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
                                gas_cost: MessageGasCost::default(),
                                execution_trace: structured::parse_events(ret.exec_trace())
                                    .unwrap_or_default(),
                            },
                            post_root,
                        ));
                    }

                    let (ret, _) = vm.apply_message(msg)?;
                    gas_reward += ret.miner_tip();
                    penalty += ret.penalty();
                }

                if let Some(rew_msg) =
                    vm.reward_message(epoch, block.miner, block.win_count, penalty, gas_reward)?
                {
                    let (ret, _) = vm.apply_implicit_message(&rew_msg)?;
                    if let Some(err) = ret.failure_info() {
                        bail!(
                            "failed to apply reward message for miner {}: {err}",
                            block.miner
                        );
                    }

                    // This is more of a sanity check, this should not be able to be hit.
                    if !ret.exit_code().is_success() {
                        bail!(
                            "reward application message failed (exit: {:?})",
                            ret.exit_code()
                        );
                    }
                }
            }

            bail!("message {target_msg_cid} not found in tipset")
        })?)
    }

    /// Validates all tipsets at epoch `start..=end` behind the heaviest tipset.
    ///
    /// Tipsets are processed sequentially. The compute-intensive work inside each
    /// tipset (`bellperson` proof verification, FVM batch seal verification, etc.)
    /// is already heavily rayon-parallelized. Parallelizing the outer loop actually introduces
    /// some issues due to locks in the aforementioned crates. So don't do it.
    ///
    /// # What is validation?
    /// Every state transition returns a new _state root_, which is typically retained in, e.g., snapshots.
    /// For "full" snapshots, all state roots are retained.
    /// For standard snapshots, the last 2000 or so state roots are retained.
    ///
    /// _receipts_ meanwhile, are typically ephemeral, but each tipset knows the _receipt root_
    /// (hash) of the previous tipset.
    ///
    /// This function takes advantage of that fact to validate tipsets:
    /// - `tipset[N]` claims that `receipt_root[N-1]` should be `0xDEADBEEF`
    /// - find `tipset[N-1]`, and perform its state transition to get the actual `receipt_root`
    /// - assert that they match
    ///
    /// See [`Self::compute_tipset_state_blocking`] for an explanation of state transitions.
    #[tracing::instrument(skip(self))]
    pub fn validate_range_blocking(&self, epochs: RangeInclusive<i64>) -> anyhow::Result<()> {
        let heaviest = self.heaviest_tipset();
        let heaviest_epoch = heaviest.epoch();
        let end = self.chain_index().load_required_tipset_by_height_blocking(
            *epochs.end(),
            heaviest,
            ResolveNullTipset::TakeOlder,
        ).with_context(|| {
            format!(
                "couldn't get a tipset at height {} behind heaviest tipset at height {heaviest_epoch}",
                *epochs.end(),
            )})?;

        // lookup tipset parents as we go along, iterating DOWN from `end`
        let tipsets = end
            .chain(self.db())
            .take_while(|ts| ts.epoch() >= *epochs.start());

        self.validate_tipsets_blocking(tipsets)
    }

    pub fn validate_tipsets_blocking<T>(&self, tipsets: T) -> anyhow::Result<()>
    where
        T: Iterator<Item = Tipset> + Send,
    {
        validate_tipsets_blocking(
            self.chain_index(),
            self.chain_config(),
            self.beacon_schedule(),
            &self.engine,
            tipsets,
        )
    }

    pub async fn execution_trace(
        &self,
        tipset: &Tipset,
        source: CallSource,
    ) -> anyhow::Result<(Cid, Vec<Arc<ApiInvocResult>>)> {
        let key = tipset.key();
        let (state_root, invoc_trace) = self
            .trace_cache
            .get_or_insert_async(
                key,
                self.execution_trace_inner(tipset.shallow_clone(), source),
            )
            .await?;
        Ok((state_root.into(), invoc_trace))
    }

    async fn execution_trace_inner(
        &self,
        tipset: Tipset,
        source: CallSource,
    ) -> anyhow::Result<(CidWrapper, Vec<Arc<ApiInvocResult>>)> {
        // Internal calls like cache prefilling should not compete the semaphore
        let permit = match source {
            CallSource::External => Some(self.replay_permit().await),
            CallSource::Internal => None,
        };
        let this = self.shallow_clone();
        tokio::task::spawn_blocking(move || {
            let _permit = permit;
            this.execution_trace_inner_blocking(tipset)
        })
        .await
        .context("tokio join error")?
    }

    fn execution_trace_inner_blocking(
        &self,
        tipset: Tipset,
    ) -> anyhow::Result<(CidWrapper, Vec<Arc<ApiInvocResult>>)> {
        let mut invoc_trace = vec![];

        let callback = |ctx: MessageCallbackCtx<'_>| {
            match ctx.at {
                CalledAt::Applied | CalledAt::Reward => {
                    invoc_trace.push(Arc::new(ApiInvocResult {
                        msg_cid: ctx.message.cid(),
                        msg: ctx.message.message().clone(),
                        msg_rct: Some(ctx.apply_ret.msg_receipt()),
                        error: ctx.apply_ret.failure_info().unwrap_or_default(),
                        duration: ctx.duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
                        gas_cost: MessageGasCost::new(ctx.message.message(), ctx.apply_ret)?,
                        execution_trace: structured::parse_events(ctx.apply_ret.exec_trace())
                            .unwrap_or_default(),
                    }));
                    Ok(())
                }
                _ => Ok(()), // ignored
            }
        };

        let ExecutedTipset { state_root, .. } = apply_block_messages_blocking(
            self.chain_index().shallow_clone(),
            self.chain_config().shallow_clone(),
            self.beacon_schedule().shallow_clone(),
            &self.engine,
            tipset,
            Some(callback),
            VMTrace::Traced,
        )?;

        Ok((state_root.into(), invoc_trace))
    }
}