forest/state_manager/
mod.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4#[cfg(test)]
5mod tests;
6
7mod actor_queries;
8mod address_resolution;
9pub mod cache;
10pub mod chain_rand;
11pub mod circulating_supply;
12mod errors;
13mod execution;
14mod message_search;
15mod message_simulation;
16mod mining;
17mod state_computation;
18pub mod utils;
19
20use self::circulating_supply::GenesisInfo;
21pub use self::errors::*;
22pub use self::message_search::FAILED_TO_LOAD_MESSAGE;
23pub use self::state_computation::{apply_block_messages_blocking, validate_tipsets_blocking};
24use crate::beacon::BeaconSchedule;
25use crate::blocks::{Tipset, TipsetKey};
26use crate::chain::{
27    ChainStore,
28    index::{ChainIndex, ResolveNullTipset},
29};
30use crate::db::DbImpl;
31use crate::interpreter::MessageCallbackCtx;
32use crate::lotus_json::{LotusJson, lotus_json_with_self};
33use crate::message::ChainMessage;
34use crate::networks::ChainConfig;
35use crate::prelude::*;
36use crate::rpc::state::ApiInvocResult;
37use crate::rpc::types::SectorOnChainInfo;
38use crate::shim::actors::init::{self, State};
39use crate::shim::actors::*;
40use crate::shim::address::AddressId;
41use crate::shim::{
42    actors::LoadActorStateFromBlockstore,
43    executor::{Receipt, StampedEvent},
44};
45use crate::shim::{
46    address::Address,
47    clock::ChainEpoch,
48    econ::TokenAmount,
49    machine::{GLOBAL_MULTI_ENGINE, MultiEngine},
50    state_tree::{ActorState, StateTree},
51    version::NetworkVersion,
52};
53use crate::state_manager::cache::ForestCache;
54use crate::utils::cache::SizeTrackingCache;
55use crate::utils::get_size::GetSize;
56use anyhow::Context as _;
57use chain_rand::ChainRand;
58use itertools::Either;
59use nonzero_ext::nonzero;
60use schemars::JsonSchema;
61use serde::{Deserialize, Serialize};
62use std::num::NonZeroUsize;
63use tracing::warn;
64
65const DEFAULT_TIPSET_CACHE_SIZE: NonZeroUsize = nonzero!(8192usize); // maximum ~150MiB on mainnet
66const DEFAULT_ID_TO_DETERMINISTIC_ADDRESS_CACHE_SIZE: NonZeroUsize = nonzero!(8192usize); // maximum ~0.7MiB on mainnet
67const DEFAULT_TRACE_CACHE_SIZE: NonZeroUsize = nonzero!(16usize); // maximum ~70MiB on mainnet
68pub const EVENTS_AMT_BITWIDTH: u32 = 5;
69pub type IdToAddressCache = SizeTrackingCache<AddressId, Address>;
70
71/// Result of executing an individual chain message in a tipset.
72///
73/// Includes the executed message itself, the execution receipt, and
74/// optional events emitted by the actor during execution.
75#[derive(Debug, Clone)]
76pub struct ExecutedMessage {
77    pub message: ChainMessage,
78    pub receipt: Receipt,
79    pub events: Option<Vec<StampedEvent>>,
80}
81
82impl GetSize for ExecutedMessage {
83    fn get_heap_size_with_tracker<T: get_size2::GetSizeTracker>(
84        &self,
85        mut tracker: T,
86    ) -> (usize, T) {
87        (
88            self.message.get_heap_size_with_tracker(&mut tracker).0
89                + self.receipt.get_heap_size_with_tracker(&mut tracker).0
90                + self.events.get_heap_size_with_tracker(&mut tracker).0,
91            tracker,
92        )
93    }
94}
95
96/// A tipset's messages paired with their execution receipts. The variants are an internal detail of
97/// how the pairs were loaded; consumers iterate them via [`Self::iter`].
98pub enum TipsetMessageReceipts {
99    /// Backed by a full executed tipset (a cache hit, or the fallback loader).
100    Executed(Arc<Vec<ExecutedMessage>>),
101    /// Receipts read directly from the child tipset, paired with the tipset's messages (no events).
102    Stored(Arc<Vec<ChainMessage>>, Vec<Receipt>),
103}
104
105impl TipsetMessageReceipts {
106    /// Iterates the (message, receipt) pairs in index order, by reference.
107    pub fn iter(&self) -> impl Iterator<Item = (&ChainMessage, &Receipt)> + '_ {
108        match self {
109            Self::Executed(messages) => {
110                Either::Left(messages.iter().map(|em| (&em.message, &em.receipt)))
111            }
112            Self::Stored(messages, receipts) => Either::Right(messages.iter().zip(receipts.iter())),
113        }
114    }
115}
116
117/// Aggregated execution result for a tipset.
118#[derive(Debug, Clone, GetSize)]
119pub struct ExecutedTipset {
120    /// Resulting state tree root after message execution
121    #[get_size(ignore)]
122    pub state_root: Cid,
123    /// Resulting message receipts root after message execution
124    #[get_size(ignore)]
125    pub receipt_root: Cid,
126    /// Per-message execution details.
127    /// Wrapped in an `Arc` to reduce cloning cost, as this can be quite large.
128    pub executed_messages: Arc<Vec<ExecutedMessage>>,
129}
130
131/// Basic execution result for a tipset.
132#[derive(Debug, Clone, GetSize)]
133pub struct TipsetState {
134    /// Resulting state tree root after message execution
135    #[get_size(ignore)]
136    pub state_root: Cid,
137    /// Resulting message receipts root after message execution
138    #[allow(dead_code)]
139    #[get_size(ignore)]
140    pub receipt_root: Cid,
141}
142
143impl From<ExecutedTipset> for TipsetState {
144    fn from(
145        ExecutedTipset {
146            state_root,
147            receipt_root,
148            ..
149        }: ExecutedTipset,
150    ) -> Self {
151        Self {
152            state_root,
153            receipt_root,
154        }
155    }
156}
157
158impl From<&ExecutedTipset> for TipsetState {
159    fn from(
160        ExecutedTipset {
161            state_root,
162            receipt_root,
163            ..
164        }: &ExecutedTipset,
165    ) -> Self {
166        Self {
167            state_root: *state_root,
168            receipt_root: *receipt_root,
169        }
170    }
171}
172
173/// External format for returning market balance from state.
174#[derive(
175    Debug, Default, Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, JsonSchema,
176)]
177#[serde(rename_all = "PascalCase")]
178pub struct MarketBalance {
179    #[schemars(with = "LotusJson<TokenAmount>")]
180    #[serde(with = "crate::lotus_json")]
181    pub escrow: TokenAmount,
182    #[schemars(with = "LotusJson<TokenAmount>")]
183    #[serde(with = "crate::lotus_json")]
184    pub locked: TokenAmount,
185}
186lotus_json_with_self!(MarketBalance);
187
188/// State manager handles all interactions with the internal Filecoin actors
189/// state. This encapsulates the [`ChainStore`] functionality, which only
190/// handles chain data, to allow for interactions with the underlying state of
191/// the chain. The state manager not only allows interfacing with state, but
192/// also is used when performing state transitions.
193pub struct StateManager {
194    /// Chain store
195    cs: ChainStore,
196    /// This is a cache which indexes tipsets to their calculated state output (state root, receipt root).
197    cache: ForestCache<TipsetKey, ExecutedTipset>,
198    /// This is a cache which indexes tipsets to their traces.
199    trace_cache: ForestCache<TipsetKey, (CidWrapper, Vec<Arc<ApiInvocResult>>)>,
200    /// `None` disables caching of ID -> deterministic-address resolution.
201    /// Used by the RPC test-snapshot generator and replay harness so every
202    /// `(id, tipset)` pair resolves independently, making recorded snapshots
203    /// read-complete and replay order-independent.
204    id_to_deterministic_address_cache: Option<IdToAddressCache>,
205    beacon: Arc<crate::beacon::BeaconSchedule>,
206    engine: Arc<MultiEngine>,
207    genesis_info: Arc<GenesisInfo>,
208    /// Bounds concurrent RPC-triggered tipset replays, see [`Self::replay_concurrency`].
209    replay_semaphore: Arc<tokio::sync::Semaphore>,
210}
211
212impl ShallowClone for StateManager {
213    fn shallow_clone(&self) -> Self {
214        Self {
215            cs: self.cs.shallow_clone(),
216            cache: self.cache.shallow_clone(),
217            trace_cache: self.trace_cache.shallow_clone(),
218            id_to_deterministic_address_cache: self
219                .id_to_deterministic_address_cache
220                .as_ref()
221                .map(ShallowClone::shallow_clone),
222            beacon: self.beacon.shallow_clone(),
223            engine: self.engine.shallow_clone(),
224            genesis_info: self.genesis_info.shallow_clone(),
225            replay_semaphore: self.replay_semaphore.shallow_clone(),
226        }
227    }
228}
229
230#[allow(clippy::type_complexity)]
231pub const NO_CALLBACK: Option<fn(MessageCallbackCtx<'_>) -> anyhow::Result<()>> = None;
232
233/// Controls whether the VM should flush its state after execution
234#[derive(Debug, Copy, Clone, Default)]
235pub enum VMFlush {
236    Flush,
237    #[default]
238    Skip,
239}
240
241/// Controls whether the FVM sender checks are enforced when simulating a message.
242#[derive(Debug, Copy, Clone, Default, PartialEq, Eq)]
243pub enum SenderValidation {
244    #[default]
245    Enforce,
246    Skip,
247}
248
249impl StateManager {
250    pub fn new(cs: ChainStore) -> anyhow::Result<Self> {
251        Self::new_with_engine(cs, GLOBAL_MULTI_ENGINE.clone())
252    }
253
254    pub fn new_with_engine(cs: ChainStore, engine: Arc<MultiEngine>) -> anyhow::Result<Self> {
255        let genesis = cs.genesis_block_header();
256        let beacon = Arc::new(cs.chain_config().get_beacon_schedule(genesis.timestamp));
257        let genesis_info = Arc::new(GenesisInfo::from_chain_config(
258            cs.chain_config().shallow_clone(),
259        ));
260
261        Ok(Self {
262            cs,
263            cache: ForestCache::new("tipset_state_executed_tipset"), // For StateOutput
264            trace_cache: ForestCache::with_size("tipset_trace", DEFAULT_TRACE_CACHE_SIZE),
265            beacon,
266            engine,
267            genesis_info,
268            id_to_deterministic_address_cache: Some(SizeTrackingCache::new_with_metrics(
269                "id_to_deterministic_address",
270                DEFAULT_ID_TO_DETERMINISTIC_ADDRESS_CACHE_SIZE,
271            )),
272            replay_semaphore: Arc::new(tokio::sync::Semaphore::new(Self::replay_concurrency())),
273        })
274    }
275
276    /// Maximum concurrent RPC-triggered tipset replays (`StateReplay`, `trace_*` and
277    /// `debug_trace*` methods). Each replay is a full VM execution of a tipset, so
278    /// unbounded concurrency lets a burst of such requests starve the whole node.
279    /// Configurable via `FOREST_RPC_REPLAY_CONCURRENCY`; defaults to half the
280    /// available CPUs.
281    fn replay_concurrency() -> usize {
282        static VALUE: std::sync::LazyLock<NonZeroUsize> = std::sync::LazyLock::new(|| {
283            let default = std::thread::available_parallelism()
284                .ok()
285                .and_then(|n| NonZeroUsize::new(n.get() / 2))
286                .unwrap_or_else(|| nonzero!(1usize));
287            crate::utils::misc::env::env_or_default("FOREST_RPC_REPLAY_CONCURRENCY", default)
288        });
289        VALUE.get()
290    }
291
292    /// The returned permit is owned so it can be moved into the `spawn_blocking`
293    /// closure doing the actual execution: if the requesting future is cancelled
294    /// (RPC timeout, disconnect), the permit stays held until the orphaned
295    /// blocking task finishes, keeping the concurrency bound accurate.
296    async fn replay_permit(&self) -> tokio::sync::OwnedSemaphorePermit {
297        self.replay_semaphore
298            .clone()
299            .acquire_owned()
300            .await
301            .expect("replay semaphore is never closed")
302    }
303
304    /// Disables caching of ID -> deterministic-address resolution. To be used strictly
305    /// by the RPC test-snapshot generator and replay harness
306    pub fn with_id_address_cache_disabled(mut self) -> Self {
307        self.id_to_deterministic_address_cache = None;
308        self
309    }
310
311    /// Test-only view of the ID -> deterministic-address cache.
312    #[cfg(test)]
313    pub(crate) fn id_to_deterministic_address_cache(&self) -> Option<&IdToAddressCache> {
314        self.id_to_deterministic_address_cache.as_ref()
315    }
316
317    /// Returns the currently tracked heaviest tipset.
318    pub fn heaviest_tipset(&self) -> Tipset {
319        self.chain_store().heaviest_tipset()
320    }
321
322    /// Returns the currently tracked heaviest tipset and rewind to a most recent valid one if necessary.
323    /// A valid head has
324    ///     - state tree in the blockstore
325    ///     - actor bundle version in the state tree that matches chain configuration
326    pub async fn maybe_rewind_heaviest_tipset(&self) -> anyhow::Result<()> {
327        while self.maybe_rewind_heaviest_tipset_once().await? {}
328        Ok(())
329    }
330
331    async fn maybe_rewind_heaviest_tipset_once(&self) -> anyhow::Result<bool> {
332        let head = self.heaviest_tipset();
333        if let Some(info) = self
334            .chain_config()
335            .network_height_with_actor_bundle(head.epoch())
336        {
337            let expected_height_info = info.info;
338            let expected_bundle = info.manifest(self.db())?;
339            let expected_bundle_metadata = expected_bundle.metadata()?;
340            let state = self.get_state_tree(head.parent_state())?;
341            let bundle_metadata = state.get_actor_bundle_metadata()?;
342            if expected_bundle_metadata != bundle_metadata {
343                let current_epoch = head.epoch();
344                let target_head = self
345                    .chain_index()
346                    .load_required_tipset_by_height(
347                        (expected_height_info.epoch - 1).max(0),
348                        head,
349                        ResolveNullTipset::TakeOlder,
350                    )
351                    .await?;
352                let target_epoch = target_head.epoch();
353                let bundle_version = &bundle_metadata.version;
354                let expected_bundle_version = &expected_bundle_metadata.version;
355                if target_epoch < current_epoch {
356                    tracing::warn!(
357                        "rewinding chain head from {current_epoch} to {target_epoch}, actor bundle: {bundle_version}, expected: {expected_bundle_version}"
358                    );
359                    if self.db().has(target_head.parent_state())? {
360                        self.chain_store().set_heaviest_tipset(target_head)?;
361                        return Ok(true);
362                    } else {
363                        anyhow::bail!(
364                            "failed to rewind, state tree @ {target_epoch} is missing from blockstore: {}",
365                            target_head.parent_state()
366                        );
367                    }
368                }
369            }
370        }
371        Ok(false)
372    }
373
374    pub fn beacon_schedule(&self) -> &Arc<BeaconSchedule> {
375        &self.beacon
376    }
377
378    /// Returns network version for the given epoch.
379    pub fn get_network_version(&self, epoch: ChainEpoch) -> NetworkVersion {
380        self.chain_config().network_version(epoch)
381    }
382
383    /// Gets the state tree
384    pub fn get_state_tree(&self, state_cid: &Cid) -> anyhow::Result<StateTree<DbImpl>> {
385        StateTree::new_from_root(self.chain_index().db(), state_cid)
386    }
387
388    /// Gets actor from given [`Cid`], if it exists.
389    pub fn get_actor(&self, addr: &Address, state_cid: Cid) -> anyhow::Result<Option<ActorState>> {
390        let state = self.get_state_tree(&state_cid)?;
391        state.get_actor(addr)
392    }
393
394    /// Gets actor state from implicit actor address
395    pub fn get_actor_state<S: LoadActorStateFromBlockstore>(
396        &self,
397        ts: &Tipset,
398    ) -> anyhow::Result<S> {
399        let state_tree = self.get_state_tree(ts.parent_state())?;
400        state_tree.get_actor_state()
401    }
402
403    /// Gets actor state from explicit actor address
404    pub fn get_actor_state_from_address<S: LoadActorStateFromBlockstore>(
405        &self,
406        ts: &Tipset,
407        actor_address: &Address,
408    ) -> anyhow::Result<S> {
409        let state_tree = self.get_state_tree(ts.parent_state())?;
410        state_tree.get_actor_state_from_address(actor_address)
411    }
412
413    /// Gets required actor from given [`Cid`].
414    pub fn get_required_actor(&self, addr: &Address, state_cid: Cid) -> anyhow::Result<ActorState> {
415        let state = self.get_state_tree(&state_cid)?;
416        state.get_actor(addr)?.with_context(|| {
417            format!("Failed to load actor with addr={addr}, state_cid={state_cid}")
418        })
419    }
420
421    /// Returns a reference to the state manager's [`Blockstore`].
422    pub fn db(&self) -> &DbImpl {
423        self.cs.db()
424    }
425
426    pub fn db_owned(&self) -> DbImpl {
427        self.cs.db_owned()
428    }
429
430    /// Returns reference to the state manager's [`ChainStore`].
431    pub fn chain_store(&self) -> &ChainStore {
432        &self.cs
433    }
434
435    /// Returns reference to the state manager's [`ChainIndex`].
436    pub fn chain_index(&self) -> &ChainIndex {
437        self.cs.chain_index()
438    }
439
440    /// Returns reference to the state manager's [`ChainConfig`].
441    pub fn chain_config(&self) -> &Arc<ChainConfig> {
442        self.cs.chain_config()
443    }
444
445    pub fn genesis_info(&self) -> &Arc<GenesisInfo> {
446        &self.genesis_info
447    }
448
449    pub fn chain_rand(&self, tipset: Tipset) -> ChainRand {
450        ChainRand::new(
451            self.chain_config().shallow_clone(),
452            tipset,
453            self.chain_index().shallow_clone(),
454            self.beacon.shallow_clone(),
455        )
456    }
457
458    /// Returns the internal, protocol-level network chain from the state.
459    pub fn get_network_state_name(
460        &self,
461        state_cid: Cid,
462    ) -> anyhow::Result<crate::networks::StateNetworkName> {
463        let init_act = self
464            .get_actor(&init::ADDRESS.into(), state_cid)?
465            .ok_or_else(|| Error::state("Init actor address could not be resolved"))?;
466        Ok(State::load(self.db(), init_act.code, init_act.state)?
467            .into_network_name()
468            .into())
469    }
470
471    /// Returns true if miner has been slashed or is considered invalid.
472    pub fn is_miner_slashed(&self, addr: &Address, state_cid: &Cid) -> anyhow::Result<bool, Error> {
473        let actor = self
474            .get_actor(&Address::POWER_ACTOR, *state_cid)?
475            .ok_or_else(|| Error::state("Power actor address could not be resolved"))?;
476
477        let spas = power::State::load(self.db(), actor.code, actor.state)?;
478
479        Ok(spas.miner_power(self.db(), addr)?.is_none())
480    }
481
482    /// Returns raw work address of a miner given the state root.
483    pub fn get_miner_work_addr(&self, state_cid: Cid, addr: &Address) -> Result<Address, Error> {
484        let state = StateTree::new_from_root(self.db(), &state_cid).map_err(Error::other)?;
485        let ms: miner::State = state.get_actor_state_from_address(addr)?;
486        let info = ms.info(self.db()).map_err(|e| e.to_string())?;
487        let addr = state.resolve_to_deterministic_address(self.db(), info.worker())?;
488        Ok(addr)
489    }
490
491    /// Returns specified actor's claimed power and total network power as a
492    /// tuple.
493    pub fn get_power(
494        &self,
495        state_cid: &Cid,
496        addr: Option<&Address>,
497    ) -> anyhow::Result<Option<(power::Claim, power::Claim)>, Error> {
498        let actor = self
499            .get_actor(&Address::POWER_ACTOR, *state_cid)?
500            .ok_or_else(|| Error::state("Power actor address could not be resolved"))?;
501
502        let spas = power::State::load(self.db(), actor.code, actor.state)?;
503
504        let t_pow = spas.total_power();
505
506        if let Some(maddr) = addr {
507            let m_pow = spas
508                .miner_power(self.db(), maddr)?
509                .ok_or_else(|| Error::state(format!("Miner for address {maddr} not found")))?;
510
511            let min_pow = spas.miner_nominal_power_meets_consensus_minimum(
512                &self.chain_config().policy,
513                self.db(),
514                maddr,
515            )?;
516            if min_pow {
517                return Ok(Some((m_pow, t_pow)));
518            }
519        }
520
521        Ok(None)
522    }
523
524    /// Single-sector lookup via a direct AMT `get`, avoiding a full sector-set scan.
525    pub fn get_sector_info(
526        &self,
527        addr: &Address,
528        sector_number: u64,
529        ts: &Tipset,
530    ) -> anyhow::Result<Option<SectorOnChainInfo>> {
531        let actor = self
532            .get_actor(addr, *ts.parent_state())?
533            .ok_or_else(|| Error::state(format!("Miner actor {addr} not found")))?;
534        let state = miner::State::load(self.db(), actor.code, actor.state)?;
535        state.get_sector(self.db(), sector_number)
536    }
537}
forest/state_manager/mod.rs

forest/state_manager/
mod.rs
