Struct MessagePool 

pub struct MessagePool<T> {
    pub(in message_pool) pending: PendingStore,
    pub(in message_pool) caches: Caches,
    pub(in message_pool) local_addrs: Arc<RwLock<HashSet<Address>>>,
    pub(in message_pool) cur_tipset: Arc<RwLock<Tipset>>,
    pub(in message_pool) api: Arc<T>,
    pub(in message_pool) network_sender: Sender<NetworkMessage>,
    pub(in message_pool) republish: Arc<RepublishState>,
    pub(in message_pool) config: Arc<MpoolConfig>,
    pub(in message_pool) chain_config: Arc<ChainConfig>,
}

This contains all necessary information needed for the message pool. Keeps track of messages to apply, as well as context needed for verifying transactions.

Fields

pending: PendingStore

Pending messages, keyed by resolved-key address, together with the broadcast channel for MpoolUpdate events. See PendingStore.

caches: Caches

local_addrs: Arc<RwLock<HashSet<Address>>>

Resolved-key senders of locally submitted messages.

cur_tipset: Arc<RwLock<Tipset>>

The current tipset (a set of blocks)

api: Arc<T>

The underlying provider

network_sender: Sender<NetworkMessage>

Sender half to send messages to other components

republish: Arc<RepublishState>

Republish coordination state

config: Arc<MpoolConfig>

Configurable parameters of the message pool.

chain_config: Arc<ChainConfig>

Chain configuration

Implementations

impl<T> MessagePool<T>

where T: Provider,

pub fn current_tipset(&self) -> Tipset

Gets the current tipset

pub(in message_pool) async fn resolve_to_key( &self, addr: &Address, cur_ts: &Tipset, ) -> Result<Address, Error>

where T: Send + Sync + 'static,

async fn add_local(&self, m: &SignedMessage) -> Result<(), Error>

where T: Send + Sync + 'static,

Record the resolved-key sender of a locally-submitted message so the republish loop can find it on its next sweep.

async fn push_internal( &self, msg: SignedMessage, trust_policy: TrustPolicy, ) -> Result<Cid, Error>

where T: Send + Sync + 'static,

Push a signed message to the MessagePool. Records the sender as local and broadcasts on gossip if validation marks it publishable.

pub(in message_pool) async fn publish_pubsub( &self, msg: &SignedMessage, ) -> Result<(), Error>

Broadcast a signed message on the network’s gossipsub topic.

pub async fn push(&self, msg: SignedMessage) -> Result<Cid, Error>

where T: Send + Sync + 'static,

Push a signed message to the MessagePool from an trusted source.

pub async fn push_untrusted(&self, msg: SignedMessage) -> Result<Cid, Error>

where T: Send + Sync + 'static,

Push a signed message to the MessagePool from an untrusted source.

pub async fn add(&self, msg: SignedMessage) -> Result<(), Error>

where T: Send + Sync + 'static,

Insert a message received via gossip. Runs full validation. Does not publish back to the network.

pub(in message_pool) async fn validate_for_pool( &self, msg: &SignedMessage, cur_ts: &Tipset, local: bool, ) -> Result<bool, Error>

where T: Send + Sync + 'static,

Message validation.

Returns publish: bool — true when the message should be gossiped after insertion; false when a local sender’s message failed the soft base-fee floor (kept locally, not broadcast).

pub(in message_pool) async fn add_to_pool( &self, msg: SignedMessage, local: bool, trust_policy: TrustPolicy, ) -> Result<bool, Error>

where T: Send + Sync + 'static,

Validate msg and insert it into the pending pool.

Returns publish: bool (see Self::validate_for_pool).

pub(in message_pool) async fn add_to_pool_unchecked( &self, cur_ts: &Tipset, msg: SignedMessage, trust_policy: TrustPolicy, strictness: StrictnessPolicy, ) -> Result<(), Error>

where T: Send + Sync + 'static,

Insert a message into the pending pool without running validation (size, sig, base-fee, sender-actor checks). The reorg replay path uses this directly to restore reverted messages even when they no longer pass the add-time filters.

pub async fn get_sequence(&self, addr: &Address) -> Result<u64, Error>

where T: Send + Sync + 'static,

Get the sequence for a given address, return Error if there is a failure to retrieve the respective sequence.

pub(in message_pool) async fn get_state_sequence( &self, addr: &Address, cur_ts: &Tipset, ) -> Result<u64, Error>

where T: Send + Sync + 'static,

Get the state nonce for an address in cur_ts, accounting for messages already included in that tipset. Cached by (TipsetKey, Address).

pub fn pending(&self) -> (Vec<SignedMessage>, Tipset)

Return a tuple that contains a vector of all signed messages and the current tipset for self.

pub async fn pending_for(&self, a: &Address) -> Option<Vec<SignedMessage>>

where T: Send + Sync + 'static,

Return a Vector of signed messages for a given from address. This vector will be sorted by each message’s sequence. If no corresponding messages found, return None result type.

pub fn subscriber(&self) -> MpoolSubscriber

A subscribe-only handle to the MpoolUpdate bus, the single entry point for observing insertions into and removals from the pending pool.

pub fn messages_for_blocks<'a>( &self, blks: impl Iterator<Item = &'a CachingBlockHeader>, ) -> Result<Vec<SignedMessage>, Error>

Return Vector of signed messages given a block header for self.

pub fn gas_limit_overestimation(&self) -> f64

pub fn config(&self) -> MpoolConfig

impl<T> MessagePool<T>

where T: Provider + Send + Sync + 'static,

pub fn new( api: T, network_sender: Sender<NetworkMessage>, config: MpoolConfig, chain_config: Arc<ChainConfig>, services: &mut JoinSet<Result<()>>, ) -> Result<Self, Error>

where T: Provider,

Creates a new MessagePool instance.

impl<T> MessagePool<T>

where T: Provider + Send + Sync + 'static,

pub(in message_pool) async fn apply_head_change( &self, revert: Vec<Tipset>, apply: Vec<Tipset>, ) -> Result<(), Error>

Revert and/or apply tipsets to the message pool.

• Apply: messages included in the new tipset are removed from the pending pool with applied = true.
• Revert: messages from the reverted tipset are re-added to the pool with StrictnessPolicy::Relaxed and TrustPolicy::Trusted, allowing them back without nonce-gap restrictions.

The state-nonce cache is naturally invalidated when the tipset changes, since it is keyed by (TipsetKey, Address).

async fn remove_applied_from_pool( &self, from: &Address, sequence: u64, rmsgs: &mut HashMap<Address, HashMap<u64, SignedMessage>>, ts: &Tipset, ) -> Result<(), Error>

Remove a message from the in-progress rmsgs scratch map. If the message isn’t there, fall back to removing it from the real pending pool. Used by Self::apply_head_change when an applied tipset includes a message that we hadn’t yet seen reverted.

impl<T: Provider> MessagePool<T>

pub(in message_pool) async fn run_republish_cycle( &self, ) -> Result<(), Error>

impl<T> MessagePool<T>

where T: Provider,

pub fn select_messages( &self, ts: &Tipset, tq: f64, ) -> Result<Vec<SignedMessage>, Error>

Forest employs a sophisticated algorithm for selecting messages for inclusion from the pool, given the ticket quality of a miner. This method selects messages for including in a block.

fn select_messages_greedy( &self, cur_ts: &Tipset, ts: &Tipset, ) -> Result<SelectedMessages, Error>

fn select_messages_optimal( &self, cur_ts: &Tipset, target_tipset: &Tipset, ticket_quality: f64, ) -> Result<SelectedMessages, Error>

fn get_pending_messages( &self, cur_ts: &Tipset, ts: &Tipset, ) -> Result<HashMap<Address, HashMap<u64, SignedMessage>>, Error>

fn select_priority_messages( &self, pending: &mut HashMap<Address, HashMap<u64, SignedMessage>>, base_fee: &TokenAmount, ts: &Tipset, ) -> Result<SelectedMessages, Error>

Trait Implementations

impl<T> ShallowClone for MessagePool<T>

fn shallow_clone(&self) -> Self

Performs a lightweight clone.