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

forest/message_pool/msgpool/
provider.rs
