forest/state_manager/

mod.rs

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

#[cfg(test)]
mod tests;

mod actor_queries;
mod address_resolution;
pub mod cache;
pub mod chain_rand;
pub mod circulating_supply;
mod errors;
mod execution;
mod message_search;
mod message_simulation;
mod mining;
mod state_computation;
pub mod utils;

pub use self::errors::*;
pub use self::state_computation::{apply_block_messages_blocking, validate_tipsets_blocking};
use crate::beacon::BeaconSchedule;
use crate::blocks::{Tipset, TipsetKey};
use crate::chain::{
    ChainStore,
    index::{ChainIndex, ResolveNullTipset},
};
use crate::db::DbImpl;
use crate::interpreter::MessageCallbackCtx;
use crate::lotus_json::{LotusJson, lotus_json_with_self};
use crate::message::ChainMessage;
use crate::networks::ChainConfig;
use crate::prelude::*;
use crate::rpc::state::ApiInvocResult;
use crate::rpc::types::SectorOnChainInfo;
use crate::shim::actors::init::{self, State};
use crate::shim::actors::*;
use crate::shim::address::AddressId;
use crate::shim::{
    actors::{LoadActorStateFromBlockstore, miner::ext::MinerStateExt as _},
    executor::{Receipt, StampedEvent},
};
use crate::shim::{
    address::Address,
    clock::ChainEpoch,
    econ::TokenAmount,
    machine::{GLOBAL_MULTI_ENGINE, MultiEngine},
    state_tree::{ActorState, StateTree},
    version::NetworkVersion,
};
use crate::state_manager::cache::ForestCache;
use crate::utils::cache::SizeTrackingCache;
use crate::utils::get_size::GetSize;
use anyhow::Context as _;
use chain_rand::ChainRand;
use nonzero_ext::nonzero;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::num::NonZeroUsize;
use tracing::warn;

const DEFAULT_TIPSET_CACHE_SIZE: NonZeroUsize = nonzero!(8192usize); // maximum ~150MiB on mainnet
const DEFAULT_ID_TO_DETERMINISTIC_ADDRESS_CACHE_SIZE: NonZeroUsize = nonzero!(8192usize); // maximum ~0.7MiB on mainnet
const DEFAULT_TRACE_CACHE_SIZE: NonZeroUsize = nonzero!(16usize); // maximum ~70MiB on mainnet
pub const EVENTS_AMT_BITWIDTH: u32 = 5;
pub type IdToAddressCache = SizeTrackingCache<AddressId, Address>;

/// Result of executing an individual chain message in a tipset.
///
/// Includes the executed message itself, the execution receipt, and
/// optional events emitted by the actor during execution.
#[derive(Debug, Clone)]
pub struct ExecutedMessage {
    pub message: ChainMessage,
    pub receipt: Receipt,
    pub events: Option<Vec<StampedEvent>>,
}

impl GetSize for ExecutedMessage {
    fn get_heap_size_with_tracker<T: get_size2::GetSizeTracker>(
        &self,
        mut tracker: T,
    ) -> (usize, T) {
        (
            self.message.get_heap_size_with_tracker(&mut tracker).0
                + self.receipt.get_heap_size_with_tracker(&mut tracker).0
                + self.events.get_heap_size_with_tracker(&mut tracker).0,
            tracker,
        )
    }
}

/// Aggregated execution result for a tipset.
#[derive(Debug, Clone, GetSize)]
pub struct ExecutedTipset {
    /// Resulting state tree root after message execution
    #[get_size(ignore)]
    pub state_root: Cid,
    /// Resulting message receipts root after message execution
    #[get_size(ignore)]
    pub receipt_root: Cid,
    /// Per-message execution details.
    /// Wrapped in an `Arc` to reduce cloning cost, as this can be quite large.
    pub executed_messages: Arc<Vec<ExecutedMessage>>,
}

/// Basic execution result for a tipset.
#[derive(Debug, Clone, GetSize)]
pub struct TipsetState {
    /// Resulting state tree root after message execution
    #[get_size(ignore)]
    pub state_root: Cid,
    /// Resulting message receipts root after message execution
    #[allow(dead_code)]
    #[get_size(ignore)]
    pub receipt_root: Cid,
}

impl From<ExecutedTipset> for TipsetState {
    fn from(
        ExecutedTipset {
            state_root,
            receipt_root,
            ..
        }: ExecutedTipset,
    ) -> Self {
        Self {
            state_root,
            receipt_root,
        }
    }
}

impl From<&ExecutedTipset> for TipsetState {
    fn from(
        ExecutedTipset {
            state_root,
            receipt_root,
            ..
        }: &ExecutedTipset,
    ) -> Self {
        Self {
            state_root: *state_root,
            receipt_root: *receipt_root,
        }
    }
}

/// External format for returning market balance from state.
#[derive(
    Debug, Default, Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, JsonSchema,
)]
#[serde(rename_all = "PascalCase")]
pub struct MarketBalance {
    #[schemars(with = "LotusJson<TokenAmount>")]
    #[serde(with = "crate::lotus_json")]
    pub escrow: TokenAmount,
    #[schemars(with = "LotusJson<TokenAmount>")]
    #[serde(with = "crate::lotus_json")]
    pub locked: TokenAmount,
}
lotus_json_with_self!(MarketBalance);

/// State manager handles all interactions with the internal Filecoin actors
/// state. This encapsulates the [`ChainStore`] functionality, which only
/// handles chain data, to allow for interactions with the underlying state of
/// the chain. The state manager not only allows interfacing with state, but
/// also is used when performing state transitions.
pub struct StateManager {
    /// Chain store
    cs: ChainStore,
    /// This is a cache which indexes tipsets to their calculated state output (state root, receipt root).
    cache: ForestCache<TipsetKey, ExecutedTipset>,
    /// This is a cache which indexes tipsets to their traces.
    trace_cache: ForestCache<TipsetKey, (CidWrapper, Vec<Arc<ApiInvocResult>>)>,
    /// `None` disables caching of ID -> deterministic-address resolution.
    /// Used by the RPC test-snapshot generator and replay harness so every
    /// `(id, tipset)` pair resolves independently, making recorded snapshots
    /// read-complete and replay order-independent.
    id_to_deterministic_address_cache: Option<IdToAddressCache>,
    beacon: Arc<crate::beacon::BeaconSchedule>,
    engine: Arc<MultiEngine>,
    /// Bounds concurrent RPC-triggered tipset replays, see [`Self::replay_concurrency`].
    replay_semaphore: Arc<tokio::sync::Semaphore>,
}

impl ShallowClone for StateManager {
    fn shallow_clone(&self) -> Self {
        Self {
            cs: self.cs.shallow_clone(),
            cache: self.cache.shallow_clone(),
            trace_cache: self.trace_cache.shallow_clone(),
            id_to_deterministic_address_cache: self
                .id_to_deterministic_address_cache
                .as_ref()
                .map(ShallowClone::shallow_clone),
            beacon: self.beacon.shallow_clone(),
            engine: self.engine.shallow_clone(),
            replay_semaphore: self.replay_semaphore.shallow_clone(),
        }
    }
}

#[allow(clippy::type_complexity)]
pub const NO_CALLBACK: Option<fn(MessageCallbackCtx<'_>) -> anyhow::Result<()>> = None;

/// Controls whether the VM should flush its state after execution
#[derive(Debug, Copy, Clone, Default)]
pub enum VMFlush {
    Flush,
    #[default]
    Skip,
}

impl StateManager {
    pub fn new(cs: ChainStore) -> anyhow::Result<Self> {
        Self::new_with_engine(cs, GLOBAL_MULTI_ENGINE.clone())
    }

    pub fn new_with_engine(cs: ChainStore, engine: Arc<MultiEngine>) -> anyhow::Result<Self> {
        let genesis = cs.genesis_block_header();
        let beacon = Arc::new(cs.chain_config().get_beacon_schedule(genesis.timestamp));

        Ok(Self {
            cs,
            cache: ForestCache::new("tipset_state_executed_tipset"), // For StateOutput
            trace_cache: ForestCache::with_size("tipset_trace", DEFAULT_TRACE_CACHE_SIZE),
            beacon,
            engine,
            id_to_deterministic_address_cache: Some(SizeTrackingCache::new_with_metrics(
                "id_to_deterministic_address",
                DEFAULT_ID_TO_DETERMINISTIC_ADDRESS_CACHE_SIZE,
            )),
            replay_semaphore: Arc::new(tokio::sync::Semaphore::new(Self::replay_concurrency())),
        })
    }

    /// Maximum concurrent RPC-triggered tipset replays (`StateReplay`, `trace_*` and
    /// `debug_trace*` methods). Each replay is a full VM execution of a tipset, so
    /// unbounded concurrency lets a burst of such requests starve the whole node.
    /// Configurable via `FOREST_RPC_REPLAY_CONCURRENCY`; defaults to half the
    /// available CPUs.
    fn replay_concurrency() -> usize {
        static VALUE: std::sync::LazyLock<NonZeroUsize> = std::sync::LazyLock::new(|| {
            let default = std::thread::available_parallelism()
                .ok()
                .and_then(|n| NonZeroUsize::new(n.get() / 2))
                .unwrap_or(nonzero!(1usize));
            crate::utils::misc::env::env_or_default("FOREST_RPC_REPLAY_CONCURRENCY", default)
        });
        VALUE.get()
    }

    /// The returned permit is owned so it can be moved into the `spawn_blocking`
    /// closure doing the actual execution: if the requesting future is cancelled
    /// (RPC timeout, disconnect), the permit stays held until the orphaned
    /// blocking task finishes, keeping the concurrency bound accurate.
    async fn replay_permit(&self) -> tokio::sync::OwnedSemaphorePermit {
        self.replay_semaphore
            .clone()
            .acquire_owned()
            .await
            .expect("replay semaphore is never closed")
    }

    /// Disables caching of ID -> deterministic-address resolution. To be used strictly
    /// by the RPC test-snapshot generator and replay harness
    pub fn with_id_address_cache_disabled(mut self) -> Self {
        self.id_to_deterministic_address_cache = None;
        self
    }

    /// Test-only view of the ID -> deterministic-address cache.
    #[cfg(test)]
    pub(crate) fn id_to_deterministic_address_cache(&self) -> Option<&IdToAddressCache> {
        self.id_to_deterministic_address_cache.as_ref()
    }

    /// Returns the currently tracked heaviest tipset.
    pub fn heaviest_tipset(&self) -> Tipset {
        self.chain_store().heaviest_tipset()
    }

    /// Returns the currently tracked heaviest tipset and rewind to a most recent valid one if necessary.
    /// A valid head has
    ///     - state tree in the blockstore
    ///     - actor bundle version in the state tree that matches chain configuration
    pub async fn maybe_rewind_heaviest_tipset(&self) -> anyhow::Result<()> {
        while self.maybe_rewind_heaviest_tipset_once().await? {}
        Ok(())
    }

    async fn maybe_rewind_heaviest_tipset_once(&self) -> anyhow::Result<bool> {
        let head = self.heaviest_tipset();
        if let Some(info) = self
            .chain_config()
            .network_height_with_actor_bundle(head.epoch())
        {
            let expected_height_info = info.info;
            let expected_bundle = info.manifest(self.db())?;
            let expected_bundle_metadata = expected_bundle.metadata()?;
            let state = self.get_state_tree(head.parent_state())?;
            let bundle_metadata = state.get_actor_bundle_metadata()?;
            if expected_bundle_metadata != bundle_metadata {
                let current_epoch = head.epoch();
                let target_head = self
                    .chain_index()
                    .load_required_tipset_by_height(
                        (expected_height_info.epoch - 1).max(0),
                        head,
                        ResolveNullTipset::TakeOlder,
                    )
                    .await?;
                let target_epoch = target_head.epoch();
                let bundle_version = &bundle_metadata.version;
                let expected_bundle_version = &expected_bundle_metadata.version;
                if target_epoch < current_epoch {
                    tracing::warn!(
                        "rewinding chain head from {current_epoch} to {target_epoch}, actor bundle: {bundle_version}, expected: {expected_bundle_version}"
                    );
                    if self.db().has(target_head.parent_state())? {
                        self.chain_store().set_heaviest_tipset(target_head)?;
                        return Ok(true);
                    } else {
                        anyhow::bail!(
                            "failed to rewind, state tree @ {target_epoch} is missing from blockstore: {}",
                            target_head.parent_state()
                        );
                    }
                }
            }
        }
        Ok(false)
    }

    pub fn beacon_schedule(&self) -> &Arc<BeaconSchedule> {
        &self.beacon
    }

    /// Returns network version for the given epoch.
    pub fn get_network_version(&self, epoch: ChainEpoch) -> NetworkVersion {
        self.chain_config().network_version(epoch)
    }

    /// Gets the state tree
    pub fn get_state_tree(&self, state_cid: &Cid) -> anyhow::Result<StateTree<DbImpl>> {
        StateTree::new_from_root(self.chain_index().db(), state_cid)
    }

    /// Gets actor from given [`Cid`], if it exists.
    pub fn get_actor(&self, addr: &Address, state_cid: Cid) -> anyhow::Result<Option<ActorState>> {
        let state = self.get_state_tree(&state_cid)?;
        state.get_actor(addr)
    }

    /// Gets actor state from implicit actor address
    pub fn get_actor_state<S: LoadActorStateFromBlockstore>(
        &self,
        ts: &Tipset,
    ) -> anyhow::Result<S> {
        let state_tree = self.get_state_tree(ts.parent_state())?;
        state_tree.get_actor_state()
    }

    /// Gets actor state from explicit actor address
    pub fn get_actor_state_from_address<S: LoadActorStateFromBlockstore>(
        &self,
        ts: &Tipset,
        actor_address: &Address,
    ) -> anyhow::Result<S> {
        let state_tree = self.get_state_tree(ts.parent_state())?;
        state_tree.get_actor_state_from_address(actor_address)
    }

    /// Gets required actor from given [`Cid`].
    pub fn get_required_actor(&self, addr: &Address, state_cid: Cid) -> anyhow::Result<ActorState> {
        let state = self.get_state_tree(&state_cid)?;
        state.get_actor(addr)?.with_context(|| {
            format!("Failed to load actor with addr={addr}, state_cid={state_cid}")
        })
    }

    /// Returns a reference to the state manager's [`Blockstore`].
    pub fn db(&self) -> &DbImpl {
        self.cs.db()
    }

    pub fn db_owned(&self) -> DbImpl {
        self.cs.db_owned()
    }

    /// Returns reference to the state manager's [`ChainStore`].
    pub fn chain_store(&self) -> &ChainStore {
        &self.cs
    }

    /// Returns reference to the state manager's [`ChainIndex`].
    pub fn chain_index(&self) -> &ChainIndex {
        self.cs.chain_index()
    }

    /// Returns reference to the state manager's [`ChainConfig`].
    pub fn chain_config(&self) -> &Arc<ChainConfig> {
        self.cs.chain_config()
    }

    pub fn chain_rand(&self, tipset: Tipset) -> ChainRand {
        ChainRand::new(
            self.chain_config().shallow_clone(),
            tipset,
            self.chain_index().shallow_clone(),
            self.beacon.shallow_clone(),
        )
    }

    /// Returns the internal, protocol-level network chain from the state.
    pub fn get_network_state_name(
        &self,
        state_cid: Cid,
    ) -> anyhow::Result<crate::networks::StateNetworkName> {
        let init_act = self
            .get_actor(&init::ADDRESS.into(), state_cid)?
            .ok_or_else(|| Error::state("Init actor address could not be resolved"))?;
        Ok(State::load(self.db(), init_act.code, init_act.state)?
            .into_network_name()
            .into())
    }

    /// Returns true if miner has been slashed or is considered invalid.
    pub fn is_miner_slashed(&self, addr: &Address, state_cid: &Cid) -> anyhow::Result<bool, Error> {
        let actor = self
            .get_actor(&Address::POWER_ACTOR, *state_cid)?
            .ok_or_else(|| Error::state("Power actor address could not be resolved"))?;

        let spas = power::State::load(self.db(), actor.code, actor.state)?;

        Ok(spas.miner_power(self.db(), addr)?.is_none())
    }

    /// Returns raw work address of a miner given the state root.
    pub fn get_miner_work_addr(&self, state_cid: Cid, addr: &Address) -> Result<Address, Error> {
        let state = StateTree::new_from_root(self.db(), &state_cid).map_err(Error::other)?;
        let ms: miner::State = state.get_actor_state_from_address(addr)?;
        let info = ms.info(self.db()).map_err(|e| e.to_string())?;
        let addr = state.resolve_to_deterministic_address(self.db(), info.worker())?;
        Ok(addr)
    }

    /// Returns specified actor's claimed power and total network power as a
    /// tuple.
    pub fn get_power(
        &self,
        state_cid: &Cid,
        addr: Option<&Address>,
    ) -> anyhow::Result<Option<(power::Claim, power::Claim)>, Error> {
        let actor = self
            .get_actor(&Address::POWER_ACTOR, *state_cid)?
            .ok_or_else(|| Error::state("Power actor address could not be resolved"))?;

        let spas = power::State::load(self.db(), actor.code, actor.state)?;

        let t_pow = spas.total_power();

        if let Some(maddr) = addr {
            let m_pow = spas
                .miner_power(self.db(), maddr)?
                .ok_or_else(|| Error::state(format!("Miner for address {maddr} not found")))?;

            let min_pow = spas.miner_nominal_power_meets_consensus_minimum(
                &self.chain_config().policy,
                self.db(),
                maddr,
            )?;
            if min_pow {
                return Ok(Some((m_pow, t_pow)));
            }
        }

        Ok(None)
    }

    // Returns all sectors
    pub fn get_all_sectors(
        &self,
        addr: &Address,
        ts: &Tipset,
    ) -> anyhow::Result<Vec<SectorOnChainInfo>> {
        let actor = self
            .get_actor(addr, *ts.parent_state())?
            .ok_or_else(|| Error::state(format!("Miner actor {addr} not found")))?;
        let state = miner::State::load(self.db(), actor.code, actor.state)?;
        state.load_sectors_ext(self.db(), None)
    }
}