forest/message_pool/msgpool/
provider.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use crate::blocks::{CachingBlockHeader, Tipset, TipsetKey};
5use crate::chain::index::ResolveNullTipset;
6use crate::chain::{ChainStore, HeadChanges};
7use crate::message::{ChainMessage, SignedMessage};
8use crate::message_pool::errors::Error;
9use crate::message_pool::msg_pool::{
10    MAX_ACTOR_PENDING_MESSAGES, MAX_UNTRUSTED_ACTOR_PENDING_MESSAGES,
11};
12use crate::networks::Height;
13use crate::prelude::*;
14use crate::shim::{
15    address::{Address, Protocol::*},
16    econ::TokenAmount,
17    message::Message,
18    state_tree::{ActorState, StateTree},
19};
20use crate::utils::db::CborStoreExt;
21use auto_impl::auto_impl;
22use tokio::sync::broadcast;
23
24/// Provider Trait. This trait will be used by the message pool to interact with
25/// some medium in order to do the operations that are listed below that are
26/// required for the message pool.
27#[auto_impl(Arc)]
28pub trait Provider {
29    /// Update `Mpool`'s `cur_tipset` whenever there is a change to the provider
30    fn subscribe_head_changes(&self) -> broadcast::Receiver<HeadChanges>;
31    /// Get the heaviest Tipset in the provider
32    fn get_heaviest_tipset(&self) -> Tipset;
33    /// Add a message to the `MpoolProvider`, return either Cid or Error
34    /// depending on successful put
35    fn put_message(&self, msg: &ChainMessage) -> Result<Cid, Error>;
36    /// Return state actor for given address given the tipset that the a temp
37    /// `StateTree` will be rooted at. Return `ActorState` or Error
38    /// depending on whether or not `ActorState` is found
39    fn get_actor_after(&self, addr: &Address, ts: &Tipset) -> Result<ActorState, Error>;
40    /// Return the signed messages for given block header
41    fn messages_for_block(
42        &self,
43        h: &CachingBlockHeader,
44    ) -> Result<(Vec<Message>, Vec<SignedMessage>), Error>;
45    /// Return a tipset given the tipset keys from the `ChainStore`
46    fn load_tipset(&self, tsk: &TipsetKey) -> Result<Tipset, Error>;
47    /// Computes the base fee
48    fn chain_compute_base_fee(&self, ts: &Tipset) -> Result<TokenAmount, Error>;
49    /// Similar to [`crate::state_manager::StateManager::resolve_to_deterministic_address`] but fails if the ID address being resolved isn't reorg-stable yet.
50    /// It should not be used for consensus-critical subsystems.
51    fn resolve_to_deterministic_address_at_finality(
52        &self,
53        addr: &Address,
54        ts: &Tipset,
55    ) -> Result<Address, Error>;
56    /// Return all messages included in the given tipset.
57    fn messages_for_tipset(&self, ts: &Tipset) -> Result<Arc<Vec<ChainMessage>>, Error>;
58    // Get max number of messages per actor in the pool
59    fn max_actor_pending_messages(&self) -> u64 {
60        MAX_ACTOR_PENDING_MESSAGES
61    }
62    // Get max number of messages per actor in the pool for untrusted sources
63    fn max_untrusted_actor_pending_messages(&self) -> u64 {
64        MAX_UNTRUSTED_ACTOR_PENDING_MESSAGES
65    }
66}
67
68impl Provider for ChainStore {
69    fn subscribe_head_changes(&self) -> broadcast::Receiver<HeadChanges> {
70        self.subscribe_head_changes()
71    }
72
73    fn get_heaviest_tipset(&self) -> Tipset {
74        self.heaviest_tipset()
75    }
76
77    fn put_message(&self, msg: &ChainMessage) -> Result<Cid, Error> {
78        let cid = self
79            .db()
80            .put_cbor_default(msg)
81            .map_err(|err| Error::Other(err.to_string()))?;
82        Ok(cid)
83    }
84
85    fn get_actor_after(&self, addr: &Address, ts: &Tipset) -> Result<ActorState, Error> {
86        let state = StateTree::new_from_root(self.db(), ts.parent_state())
87            .map_err(|e| Error::Other(e.to_string()))?;
88        Ok(state.get_required_actor(addr)?)
89    }
90
91    fn messages_for_block(
92        &self,
93        h: &CachingBlockHeader,
94    ) -> Result<(Vec<Message>, Vec<SignedMessage>), Error> {
95        crate::chain::block_messages(self.db(), h).map_err(|err| err.into())
96    }
97
98    fn load_tipset(&self, tsk: &TipsetKey) -> Result<Tipset, Error> {
99        Ok(self.chain_index().load_required_tipset(tsk)?)
100    }
101
102    fn chain_compute_base_fee(&self, ts: &Tipset) -> Result<TokenAmount, Error> {
103        let smoke_height = self.chain_config().epoch(Height::Smoke);
104        let firehorse_height = self.chain_config().epoch(Height::FireHorse);
105        crate::chain::compute_base_fee(self.db(), ts, smoke_height, firehorse_height)
106            .map_err(|err| err.into())
107    }
108
109    fn resolve_to_deterministic_address_at_finality(
110        &self,
111        addr: &Address,
112        ts: &Tipset,
113    ) -> Result<Address, Error> {
114        match addr.protocol() {
115            BLS | Secp256k1 | Delegated => Ok(*addr),
116            Actor => Err(Error::Other(
117                "Cannot resolve actor address to key address".into(),
118            )),
119            _ => {
120                let lookback_ts = if ts.epoch() > self.chain_config().policy.chain_finality {
121                    self.chain_index()
122                        .load_required_tipset_by_height_blocking(
123                            ts.epoch() - self.chain_config().policy.chain_finality,
124                            ts.clone(),
125                            ResolveNullTipset::TakeOlder,
126                        )
127                        .map_err(|e| Error::Other(e.to_string()))?
128                } else {
129                    // Matches the logic at <https://github.com/filecoin-project/lotus/blob/v1.35.1/chain/stmgr/stmgr.go#L361>
130                    ts.clone()
131                };
132
133                let state = StateTree::new_from_root(self.db(), lookback_ts.parent_state())
134                    .map_err(|e| Error::Other(e.to_string()))?;
135                state
136                    .resolve_to_deterministic_address(self.db(), *addr)
137                    .map_err(|e| Error::Other(e.to_string()))
138            }
139        }
140    }
141
142    fn messages_for_tipset(&self, ts: &Tipset) -> Result<Arc<Vec<ChainMessage>>, Error> {
143        ChainStore::messages_for_tipset(self, ts).map_err(Into::into)
144    }
145}
146
147pub trait ProviderExt {
148    /// Non-blocking version of [`Provider::resolve_to_deterministic_address_at_finality`]
149    async fn resolve_to_deterministic_address_at_finality_async(
150        &self,
151        addr: Address,
152        ts: Tipset,
153    ) -> Result<Address, Error>;
154}
155
156impl<T> ProviderExt for T
157where
158    T: Provider + ShallowClone + Send + Sync + 'static,
159{
160    async fn resolve_to_deterministic_address_at_finality_async(
161        &self,
162        addr: Address,
163        ts: Tipset,
164    ) -> Result<Address, Error> {
165        let this = self.shallow_clone();
166        tokio::task::spawn_blocking(move || {
167            this.resolve_to_deterministic_address_at_finality(&addr, &ts)
168        })
169        .await
170        .context("tokio join error")?
171    }
172}
forest/message_pool/msgpool/provider.rs

forest/message_pool/msgpool/
provider.rs
