forest/state_manager/
message_simulation.rs

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

forest/state_manager/
message_simulation.rs
