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

forest/message_pool/msgpool/
provider.rs
