forest/state_manager/
execution.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use super::state_computation::{
5    TipsetExecutor, apply_block_messages_blocking, validate_tipsets_blocking,
6};
7use super::utils::structured;
8use super::*;
9use crate::interpreter::{CalledAt, VMTrace};
10use crate::rpc::state::{ApiInvocResult, MessageGasCost};
11use anyhow::{Context as _, bail};
12use num_traits::identities::Zero;
13use std::ops::RangeInclusive;
14
15impl StateManager {
16    /// Replays the given message and returns the result of executing the
17    /// indicated message, assuming it was executed in the indicated tipset.
18    ///
19    /// Served from the shared tipset trace cache, so concurrent replays of
20    /// messages in the same tipset share a single traced execution. Unlike
21    /// Lotus, which halts at the target message, this executes the whole
22    /// tipset — the coalescing depends on it, don't port the halt back.
23    /// Consequently, failures after the target message also fail the replay.
24    pub async fn replay(&self, ts: Tipset, mcid: Cid) -> Result<ApiInvocResult, Error> {
25        let (_, trace) = self
26            .execution_trace(&ts)
27            .await
28            .map_err(|e| Error::Other(format!("unexpected error during execution : {e}")))?;
29        trace
30            .iter()
31            .find(|r| r.msg_cid == mcid)
32            .map(|r| (**r).clone())
33            .ok_or_else(|| Error::Other("failed to replay".into()))
34    }
35
36    /// Replays a tipset up to a target message, capturing the state root before
37    /// and after execution.
38    pub async fn replay_for_prestate(
39        &self,
40        ts: Tipset,
41        target_message_cid: Cid,
42    ) -> Result<(Cid, ApiInvocResult, Cid), Error> {
43        let permit = self.replay_permit().await;
44        let this = self.shallow_clone();
45        tokio::task::spawn_blocking(move || {
46            let _permit = permit;
47            this.replay_for_prestate_blocking(ts, target_message_cid)
48        })
49        .await
50        .map_err(|e| Error::Other(format!("{e}")))?
51    }
52
53    fn replay_for_prestate_blocking(
54        &self,
55        ts: Tipset,
56        target_msg_cid: Cid,
57    ) -> Result<(Cid, ApiInvocResult, Cid), Error> {
58        if ts.epoch() == 0 {
59            return Err(Error::Other(
60                "cannot trace messages in the genesis block".into(),
61            ));
62        }
63
64        let genesis_timestamp = self.chain_store().genesis_block_header().timestamp;
65        let exec = TipsetExecutor::new(
66            self.chain_index().shallow_clone(),
67            self.chain_config().shallow_clone(),
68            self.beacon_schedule().shallow_clone(),
69            &self.engine,
70            ts.shallow_clone(),
71        );
72        let mut no_cb = NO_CALLBACK;
73        let (parent_state, epoch, block_messages) =
74            exec.prepare_parent_state_blocking(genesis_timestamp, VMTrace::NotTraced, &mut no_cb)?;
75
76        Ok(stacker::grow(64 << 20, || {
77            let mut vm =
78                exec.create_vm(parent_state, epoch, ts.min_timestamp(), VMTrace::NotTraced)?;
79            let mut processed = ahash::HashSet::default();
80
81            for block in block_messages.iter() {
82                let mut penalty = TokenAmount::zero();
83                let mut gas_reward = TokenAmount::zero();
84
85                for msg in block.messages.iter() {
86                    let cid = msg.cid();
87                    if processed.contains(&cid) {
88                        continue;
89                    }
90
91                    processed.insert(cid);
92
93                    if cid == target_msg_cid {
94                        let pre_root = vm.flush()?;
95                        let mut traced_vm =
96                            exec.create_vm(pre_root, epoch, ts.min_timestamp(), VMTrace::Traced)?;
97                        let (ret, duration) = traced_vm.apply_message(msg)?;
98                        let post_root = traced_vm.flush()?;
99
100                        return Ok((
101                            pre_root,
102                            ApiInvocResult {
103                                msg_cid: cid,
104                                msg: msg.message().clone(),
105                                msg_rct: Some(ret.msg_receipt()),
106                                error: ret.failure_info().unwrap_or_default(),
107                                duration: duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
108                                gas_cost: MessageGasCost::default(),
109                                execution_trace: structured::parse_events(ret.exec_trace())
110                                    .unwrap_or_default(),
111                            },
112                            post_root,
113                        ));
114                    }
115
116                    let (ret, _) = vm.apply_message(msg)?;
117                    gas_reward += ret.miner_tip();
118                    penalty += ret.penalty();
119                }
120
121                if let Some(rew_msg) =
122                    vm.reward_message(epoch, block.miner, block.win_count, penalty, gas_reward)?
123                {
124                    let (ret, _) = vm.apply_implicit_message(&rew_msg)?;
125                    if let Some(err) = ret.failure_info() {
126                        bail!(
127                            "failed to apply reward message for miner {}: {err}",
128                            block.miner
129                        );
130                    }
131
132                    // This is more of a sanity check, this should not be able to be hit.
133                    if !ret.msg_receipt().exit_code().is_success() {
134                        bail!(
135                            "reward application message failed (exit: {:?})",
136                            ret.msg_receipt().exit_code()
137                        );
138                    }
139                }
140            }
141
142            bail!("message {target_msg_cid} not found in tipset")
143        })?)
144    }
145
146    /// Validates all tipsets at epoch `start..=end` behind the heaviest tipset.
147    ///
148    /// Tipsets are processed sequentially. The compute-intensive work inside each
149    /// tipset (`bellperson` proof verification, FVM batch seal verification, etc.)
150    /// is already heavily rayon-parallelized. Parallelizing the outer loop actually introduces
151    /// some issues due to locks in the aforementioned crates. So don't do it.
152    ///
153    /// # What is validation?
154    /// Every state transition returns a new _state root_, which is typically retained in, e.g., snapshots.
155    /// For "full" snapshots, all state roots are retained.
156    /// For standard snapshots, the last 2000 or so state roots are retained.
157    ///
158    /// _receipts_ meanwhile, are typically ephemeral, but each tipset knows the _receipt root_
159    /// (hash) of the previous tipset.
160    ///
161    /// This function takes advantage of that fact to validate tipsets:
162    /// - `tipset[N]` claims that `receipt_root[N-1]` should be `0xDEADBEEF`
163    /// - find `tipset[N-1]`, and perform its state transition to get the actual `receipt_root`
164    /// - assert that they match
165    ///
166    /// See [`Self::compute_tipset_state_blocking`] for an explanation of state transitions.
167    #[tracing::instrument(skip(self))]
168    pub fn validate_range_blocking(&self, epochs: RangeInclusive<i64>) -> anyhow::Result<()> {
169        let heaviest = self.heaviest_tipset();
170        let heaviest_epoch = heaviest.epoch();
171        let end = self.chain_index().load_required_tipset_by_height_blocking(
172            *epochs.end(),
173            heaviest,
174            ResolveNullTipset::TakeOlder,
175        ).with_context(|| {
176            format!(
177                "couldn't get a tipset at height {} behind heaviest tipset at height {heaviest_epoch}",
178                *epochs.end(),
179            )})?;
180
181        // lookup tipset parents as we go along, iterating DOWN from `end`
182        let tipsets = end
183            .chain(self.db())
184            .take_while(|ts| ts.epoch() >= *epochs.start());
185
186        self.validate_tipsets_blocking(tipsets)
187    }
188
189    pub fn validate_tipsets_blocking<T>(&self, tipsets: T) -> anyhow::Result<()>
190    where
191        T: Iterator<Item = Tipset> + Send,
192    {
193        validate_tipsets_blocking(
194            self.chain_index(),
195            self.chain_config(),
196            self.beacon_schedule(),
197            &self.engine,
198            tipsets,
199        )
200    }
201
202    pub async fn execution_trace(
203        &self,
204        tipset: &Tipset,
205    ) -> anyhow::Result<(Cid, Vec<Arc<ApiInvocResult>>)> {
206        let key = tipset.key();
207        let (state_root, invoc_trace) = self
208            .trace_cache
209            .get_or_insert_async(key, self.execution_trace_inner(tipset.shallow_clone()))
210            .await?;
211        Ok((state_root.into(), invoc_trace))
212    }
213
214    async fn execution_trace_inner(
215        &self,
216        tipset: Tipset,
217    ) -> anyhow::Result<(CidWrapper, Vec<Arc<ApiInvocResult>>)> {
218        let permit = self.replay_permit().await;
219        let this = self.shallow_clone();
220        tokio::task::spawn_blocking(move || {
221            let _permit = permit;
222            this.execution_trace_inner_blocking(tipset)
223        })
224        .await
225        .context("tokio join error")?
226    }
227
228    fn execution_trace_inner_blocking(
229        &self,
230        tipset: Tipset,
231    ) -> anyhow::Result<(CidWrapper, Vec<Arc<ApiInvocResult>>)> {
232        let mut invoc_trace = vec![];
233
234        let callback = |ctx: MessageCallbackCtx<'_>| {
235            match ctx.at {
236                CalledAt::Applied | CalledAt::Reward => {
237                    invoc_trace.push(Arc::new(ApiInvocResult {
238                        msg_cid: ctx.message.cid(),
239                        msg: ctx.message.message().clone(),
240                        msg_rct: Some(ctx.apply_ret.msg_receipt()),
241                        error: ctx.apply_ret.failure_info().unwrap_or_default(),
242                        duration: ctx.duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
243                        gas_cost: MessageGasCost::new(ctx.message.message(), ctx.apply_ret)?,
244                        execution_trace: structured::parse_events(ctx.apply_ret.exec_trace())
245                            .unwrap_or_default(),
246                    }));
247                    Ok(())
248                }
249                _ => Ok(()), // ignored
250            }
251        };
252
253        let ExecutedTipset { state_root, .. } = apply_block_messages_blocking(
254            self.chain_index().shallow_clone(),
255            self.chain_config().shallow_clone(),
256            self.beacon_schedule().shallow_clone(),
257            &self.engine,
258            tipset,
259            Some(callback),
260            VMTrace::Traced,
261        )?;
262
263        Ok((state_root.into(), invoc_trace))
264    }
265}
forest/state_manager/execution.rs

forest/state_manager/
execution.rs
