forest/state_manager/
message_simulation.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use super::circulating_supply::GenesisInfo;
5use super::utils::structured;
6use super::*;
7use crate::interpreter::{ExecutionContext, IMPLICIT_MESSAGE_GAS_LIMIT, VM, VMTrace};
8use crate::message::{MessageRead as _, MessageReadWrite as _};
9use crate::rpc::state::{ApiInvocResult, InvocResult, MessageGasCost};
10use crate::shim::executor::ApplyRet;
11use crate::shim::message::Message;
12use crate::state_migration::run_state_migrations;
13use std::time::Duration;
14use tracing::instrument;
15
16impl StateManager {
17    #[instrument(skip(self))]
18    fn call_raw_blocking(
19        &self,
20        state_cid: Option<Cid>,
21        msg: &Message,
22        tipset: Option<Tipset>,
23    ) -> Result<ApiInvocResult, Error> {
24        let mut msg = msg.clone();
25        let chain_config = self.chain_config();
26
27        let tipset = if let Some(ts) = tipset {
28            if ts.epoch() > 0 {
29                // Explicit-state calls already hold every fork below the tipset epoch, so only a
30                // migration at the tipset epoch itself is refused. Parent-state calls (no explicit
31                // state) lag a tipset, so a migration at the parent epoch must also be refused.
32                let (fork_floor, fork_height) = if state_cid.is_some() {
33                    (ts.epoch(), ts.epoch() + 1)
34                } else {
35                    let parent = self
36                        .chain_index()
37                        .load_required_tipset(ts.parents())
38                        .map_err(Error::other)?;
39                    (parent.epoch(), ts.epoch() + 1)
40                };
41                if let Some(epoch) = chain_config.expensive_fork_between(fork_floor, fork_height) {
42                    return Err(Error::ExpensiveFork { epoch });
43                }
44            }
45            ts
46        } else {
47            // Search back till we find a height with no fork, or we reach the beginning.
48            let mut heaviest_ts = self.heaviest_tipset();
49            while heaviest_ts.epoch() > 0 {
50                let parent = self
51                    .chain_index()
52                    .load_required_tipset(heaviest_ts.parents())
53                    .map_err(Error::other)?;
54                if !chain_config.has_expensive_fork_between(parent.epoch(), heaviest_ts.epoch() + 1)
55                {
56                    break;
57                }
58                heaviest_ts = parent;
59            }
60            heaviest_ts
61        };
62
63        let state_cid = state_cid.unwrap_or(*tipset.parent_state());
64
65        // Handle state forks
66        let state_cid = match run_state_migrations(
67            tipset.epoch(),
68            self.chain_config(),
69            self.db(),
70            &state_cid,
71        ) {
72            Ok(Some(new_state)) => new_state,
73            Ok(None) => state_cid,
74            Err(e) => return Err(Error::other(e)),
75        };
76
77        let height = tipset.epoch();
78        let genesis_info = GenesisInfo::from_chain_config(self.chain_config().clone());
79        let mut vm = VM::new(
80            ExecutionContext {
81                heaviest_tipset: tipset.shallow_clone(),
82                state_tree_root: state_cid,
83                epoch: height,
84                rand: Box::new(self.chain_rand(tipset.shallow_clone())),
85                base_fee: tipset.block_headers().first().parent_base_fee.clone(),
86                circ_supply: genesis_info.get_vm_circulating_supply(
87                    height,
88                    self.db(),
89                    &state_cid,
90                )?,
91                chain_config: self.chain_config().shallow_clone(),
92                chain_index: self.chain_index().shallow_clone(),
93                timestamp: tipset.min_timestamp(),
94            },
95            &self.engine,
96            VMTrace::Traced,
97        )?;
98
99        let tipset_messages = self
100            .chain_store()
101            .messages_for_tipset(&tipset)
102            .map_err(|err| Error::Other(err.to_string()))?;
103
104        let prior_messsages = tipset_messages
105            .iter()
106            .filter(|ts_msg| ts_msg.message().from() == msg.from());
107
108        for m in prior_messsages {
109            vm.apply_message(m)?;
110        }
111
112        // We flush to get the VM's view of the state tree after applying the above messages
113        // This is needed to get the correct nonce from the actor state to match the VM
114        let state_cid = vm.flush()?;
115
116        let state = StateTree::new_from_root(self.db(), &state_cid)?;
117
118        let from_actor = state
119            .get_actor(&msg.from())?
120            .ok_or_else(|| anyhow::anyhow!("actor not found"))?;
121        msg.set_sequence(from_actor.sequence);
122
123        // Implicit messages need to set a special gas limit
124        let mut msg = msg.clone();
125        msg.gas_limit = IMPLICIT_MESSAGE_GAS_LIMIT as u64;
126
127        let (apply_ret, duration) = vm.apply_implicit_message(&msg)?;
128
129        Ok(ApiInvocResult {
130            msg: msg.clone(),
131            msg_rct: Some(apply_ret.msg_receipt()),
132            msg_cid: msg.cid(),
133            error: apply_ret.failure_info().unwrap_or_default(),
134            duration: duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
135            gas_cost: MessageGasCost::default(),
136            execution_trace: structured::parse_events(apply_ret.exec_trace()).unwrap_or_default(),
137        })
138    }
139
140    /// runs the given message and returns its result without any persisted
141    /// changes.
142    pub async fn call(
143        &self,
144        message: Arc<Message>,
145        tipset: Option<Tipset>,
146    ) -> Result<ApiInvocResult, Error> {
147        let this = self.shallow_clone();
148        tokio::task::spawn_blocking(move || this.call_blocking(&message, tipset)).await?
149    }
150
151    /// Blocking version of [`Self::call`], use with caution.
152    pub fn call_blocking(
153        &self,
154        message: &Message,
155        tipset: Option<Tipset>,
156    ) -> Result<ApiInvocResult, Error> {
157        self.call_raw_blocking(None, message, tipset)
158    }
159
160    /// Same as [`StateManager::call`] but runs the message on the given state and not
161    /// on the parent state of the tipset.
162    pub async fn call_on_state(
163        &self,
164        state_cid: Cid,
165        message: Arc<Message>,
166        tipset: Option<Tipset>,
167    ) -> Result<ApiInvocResult, Error> {
168        let this = self.shallow_clone();
169        tokio::task::spawn_blocking(move || {
170            this.call_on_state_blocking(state_cid, &message, tipset)
171        })
172        .await?
173    }
174
175    /// Blocking version of [`Self::call_on_state`], use with caution.
176    pub fn call_on_state_blocking(
177        &self,
178        state_cid: Cid,
179        message: &Message,
180        tipset: Option<Tipset>,
181    ) -> Result<ApiInvocResult, Error> {
182        self.call_raw_blocking(Some(state_cid), message, tipset)
183    }
184
185    pub async fn apply_on_state_with_gas(
186        &self,
187        tipset: Option<Tipset>,
188        msg: Message,
189        vm_flush: VMFlush,
190    ) -> anyhow::Result<(ApiInvocResult, Option<Cid>)> {
191        let ts = tipset.unwrap_or_else(|| self.heaviest_tipset());
192
193        let from_a = self.resolve_to_deterministic_address(msg.from, &ts).await?;
194        let chain_msg = ChainMessage::for_gas_estimation(msg.clone(), from_a.protocol());
195
196        let (_invoc_res, apply_ret, duration, state_root) = self
197            .call_with_gas(chain_msg, Default::default(), Some(ts), vm_flush)
198            .await?;
199
200        Ok((
201            ApiInvocResult {
202                msg_cid: msg.cid(),
203                msg,
204                msg_rct: Some(apply_ret.msg_receipt()),
205                error: apply_ret.failure_info().unwrap_or_default(),
206                duration: duration.as_nanos().clamp(0, u128::from(u64::MAX)) as u64,
207                gas_cost: MessageGasCost::default(),
208                execution_trace: structured::parse_events(apply_ret.exec_trace())
209                    .unwrap_or_default(),
210            },
211            state_root,
212        ))
213    }
214
215    /// Computes message on the given [Tipset] state, after applying other
216    /// messages and returns the values computed in the VM.
217    pub async fn call_with_gas(
218        &self,
219        mut message: ChainMessage,
220        prior_messages: Arc<Vec<ChainMessage>>,
221        tipset: Option<Tipset>,
222        vm_flush: VMFlush,
223    ) -> Result<(InvocResult, ApplyRet, Duration, Option<Cid>), Error> {
224        let ts = tipset.unwrap_or_else(|| self.heaviest_tipset());
225        let TipsetState { state_root, .. } = self
226            .load_tipset_state(&ts)
227            .await
228            .map_err(|e| Error::Other(format!("Could not load tipset state: {e:#}")))?;
229        let chain_rand = self.chain_rand(ts.clone());
230
231        // Since we're simulating a future message, pretend we're applying it in the
232        // "next" tipset
233        let epoch = ts.epoch() + 1;
234        let genesis_info = GenesisInfo::from_chain_config(self.chain_config().clone());
235        let this = self.shallow_clone();
236        tokio::task::spawn_blocking(move || {
237            // FVM requires a stack size of 64MiB. The alternative is to use `ThreadedExecutor` from
238            // FVM, but that introduces some constraints, and possible deadlocks.
239            let (ret, duration, state_cid) = stacker::grow(64 << 20, || -> anyhow::Result<_> {
240                let mut vm = VM::new(
241                    ExecutionContext {
242                        heaviest_tipset: ts.clone(),
243                        state_tree_root: state_root,
244                        epoch,
245                        rand: Box::new(chain_rand),
246                        base_fee: ts.block_headers().first().parent_base_fee.clone(),
247                        circ_supply: genesis_info.get_vm_circulating_supply(
248                            epoch,
249                            this.chain_index().db(),
250                            &state_root,
251                        )?,
252                        chain_config: this.chain_config().shallow_clone(),
253                        chain_index: this.chain_index().shallow_clone(),
254                        timestamp: ts.min_timestamp(),
255                    },
256                    &this.engine,
257                    VMTrace::NotTraced,
258                )?;
259
260                for msg in prior_messages.iter() {
261                    vm.apply_message(msg)?;
262                }
263
264                let from_actor = vm
265                    .get_actor(&message.from())
266                    .map_err(|e| Error::Other(format!("Could not get actor from state: {e:#}")))?
267                    .ok_or_else(|| Error::Other("cant find actor in state tree".to_string()))?;
268
269                message.set_sequence(from_actor.sequence);
270                let (ret, duration) = vm.apply_message(&message)?;
271                let state_root = match vm_flush {
272                    VMFlush::Flush => Some(vm.flush()?),
273                    VMFlush::Skip => None,
274                };
275                Ok((ret, duration, state_root))
276            })?;
277
278            Ok((
279                InvocResult::new(message.message().clone(), &ret),
280                ret,
281                duration,
282                state_cid,
283            ))
284        })
285        .await?
286    }
287}
forest/state_manager/message_simulation.rs

forest/state_manager/
message_simulation.rs
