From 2d638163dfb3653f707a12200747e08ba6243417 Mon Sep 17 00:00:00 2001 From: Aryan Tikarya Date: Tue, 16 Jun 2026 21:15:52 +0530 Subject: [PATCH] fix the run head change in select message to only simulate the head change --- CHANGELOG.md | 2 + src/message_pool/msgpool/selection.rs | 145 ++++++++++++++++---------- src/message_pool/msgpool/utils.rs | 10 ++ 3 files changed, 104 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a448b5cba52a..55370e364cb5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,8 @@ - [#6974](https://github.com/ChainSafe/forest/issues/6974): Fixed the message pool reporting a still-pending nonce as the next nonce after an applied message was removed. +- [#6975](https://github.com/ChainSafe/forest/issues/6975): Fixed `Filecoin.MpoolSelect` to not remove the messages from the live pool, only simulate the head change. + ## Forest v0.33.6 "Ebb" Non-mandatory release for all node operators. It fixes a critical memory leak in `v0.33.5`. (Earlier releases are not affected) diff --git a/src/message_pool/msgpool/selection.rs b/src/message_pool/msgpool/selection.rs index 66b8b6a345e7..568b2e7856ad 100644 --- a/src/message_pool/msgpool/selection.rs +++ b/src/message_pool/msgpool/selection.rs @@ -15,13 +15,11 @@ use crate::message_pool::msg_chain::MsgChainNode; use crate::message_pool::{ Error, msg_chain::{Chains, NodeKey, create_message_chains}, - msg_pool::resolve_to_key, - msgpool::{MIN_GAS, pending_store::PendingStore}, + msgpool::MIN_GAS, }; use crate::prelude::*; use crate::shim::crypto::{Signature, SignatureType}; use crate::shim::{address::Address, econ::TokenAmount}; -use crate::state_manager::IdToAddressCache; use crate::utils::cache::SizeTrackingCache; use ahash::HashMap; use anyhow::{bail, ensure}; @@ -651,8 +649,6 @@ where run_head_change( self.api.as_ref(), &self.caches.bls_sig, - &self.pending, - &self.caches.key, cur_ts.clone(), ts.clone(), &mut result, @@ -797,15 +793,12 @@ fn merge_and_trim( selected_msgs } -/// Like `head_change`, except it doesn't change the state of the `MessagePool`. -/// It simulates a head change call. +/// Like `head_change`, except it simulates a head change call and doesn't change the state of the `MessagePool`. // This logic should probably be implemented in the ChainStore. It handles // reorgs. pub(in crate::message_pool) fn run_head_change( api: &T, bls_sig_cache: &SizeTrackingCache, - pending_store: &PendingStore, - key_cache: &IdToAddressCache, from: Tipset, to: Tipset, rmsgs: &mut HashMap>, @@ -852,58 +845,16 @@ where let (msgs, smsgs) = api.messages_for_block(b)?; for msg in smsgs { - remove_applied_from_pool( - api, - key_cache, - pending_store, - &ts, - &msg.from(), - msg.sequence(), - rmsgs, - )?; + utils::remove_from_selected_msgs(&msg.from(), msg.sequence(), rmsgs); } for msg in msgs { - remove_applied_from_pool( - api, - key_cache, - pending_store, - &ts, - &msg.from, - msg.sequence, - rmsgs, - )?; + utils::remove_from_selected_msgs(&msg.from, msg.sequence, rmsgs); } } } Ok(()) } -/// Free-fn mirror of [`MessagePool::remove_applied_from_pool`] for the -/// simulator path, which has only the individual fields to hand and not a -/// `&MessagePool`. Bodies are intentionally identical; consolidation can -/// happen once the simulator routes through `&MessagePool` directly. -#[allow(clippy::too_many_arguments)] -fn remove_applied_from_pool( - api: &T, - key_cache: &IdToAddressCache, - pending_store: &PendingStore, - ts: &Tipset, - from: &Address, - sequence: u64, - rmsgs: &mut HashMap>, -) -> Result<(), Error> { - if rmsgs - .get_mut(from) - .and_then(|temp| temp.remove(&sequence)) - .is_none() - && let Ok(resolved) = resolve_to_key(api, key_cache, from, ts) - .inspect_err(|e| tracing::debug!(%from, "remove: failed to resolve address: {e:#}")) - { - let _ = pending_store.remove(&resolved, sequence, true); - } - Ok(()) -} - #[cfg(test)] mod test_selection { use std::sync::Arc; @@ -1080,6 +1031,94 @@ mod test_selection { } } + #[tokio::test] + async fn select_messages_on_non_current_tipset_should_not_update_mpool_state() { + use crate::message_pool::msgpool::msg_pool::TrustPolicy; + use crate::message_pool::msgpool::msg_set::StrictnessPolicy; + use crate::shim::message::Message as ShimMessage; + use tokio::sync::broadcast::error::TryRecvError; + + let mut joinset = JoinSet::new(); + let mpool = make_test_mpool(&mut joinset); + + let id_addr = Address::new_id(1000); + let key_addr = Address::new_bls(&[3u8; 48]).unwrap(); + mpool.api.set_key_address_mapping(&id_addr, &key_addr); + mpool + .api + .set_state_balance_raw(&key_addr, TokenAmount::from_whole(1)); + mpool.api.set_state_sequence(&key_addr, 0); + + // Establish a current head registered with the provider so the + // simulated head change can walk parents back to it. + let b1 = mpool.api.next_block(); + mpool.api.set_block_messages(&b1, vec![]); + let head = Tipset::from(&b1); + mpool + .apply_head_change(Vec::new(), vec![head.clone()]) + .await + .unwrap(); + + let pending_msg = SignedMessage::mock_bls_signed_message(ShimMessage { + from: id_addr, + sequence: 0, + gas_limit: TEST_GAS_LIMIT as u64, + gas_fee_cap: TokenAmount::from_atto(200), + gas_premium: TokenAmount::from_atto(100), + ..ShimMessage::default() + }); + mpool + .add_to_pool_unchecked( + &head, + pending_msg.clone(), + TrustPolicy::Trusted, + StrictnessPolicy::Relaxed, + ) + .unwrap(); + + let before = mpool.pending.snapshot(); + assert!( + before.contains_key(&key_addr), + "precondition: message is pending under the resolved key address" + ); + + // A *non-current* child tipset whose block applies that very message, + // carrying the `f0` `from` as on-chain messages do. + let b2 = mpool.api.next_block(); + let ts2 = Tipset::from(&b2); + mpool.api.set_block_messages(&b2, vec![pending_msg.clone()]); + + // Subscribe AFTER the insert so we only observe events emitted by the + // selection call below. + let mut rx = mpool.pending.subscribe(); + + // Select against the non-current tipset. + let _ = mpool.select_messages(&ts2, 1.0).unwrap(); + + let after = mpool.pending.snapshot(); + assert!( + after.contains_key(&key_addr), + "selecting for a non-current tipset must not remove live pending messages" + ); + assert_eq!( + before.len(), + after.len(), + "the live pending pool size must be unchanged by selection" + ); + assert_eq!( + after + .get(&key_addr) + .and_then(|mset| mset.msgs.get(&0)) + .map(|m| m.cid()), + Some(pending_msg.cid()), + "the exact pending message must survive at its nonce" + ); + assert!( + matches!(rx.try_recv(), Err(TryRecvError::Empty)), + "a read-only selection simulation must not emit any MpoolUpdate events" + ); + } + #[tokio::test] async fn message_selection_trimming_gas() { let mut joinset = JoinSet::new(); diff --git a/src/message_pool/msgpool/utils.rs b/src/message_pool/msgpool/utils.rs index 4de9aa17a0ee..812773055d62 100644 --- a/src/message_pool/msgpool/utils.rs +++ b/src/message_pool/msgpool/utils.rs @@ -67,6 +67,16 @@ pub(in crate::message_pool) fn add_to_selected_msgs( rmsgs.entry(m.from()).or_default().insert(m.sequence(), m); } +pub(in crate::message_pool) fn remove_from_selected_msgs( + from: &Address, + sequence: u64, + rmsgs: &mut HashMap>, +) { + if let Some(set) = rmsgs.get_mut(from) { + set.remove(&sequence); + } +} + /// Computes the minimum gas premium required to replace an existing message /// using [`REPLACE_BY_FEE_RATIO_MIN`]. ///