From 64858146955af3417e4650eebd7e0bab12f02654 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Wed, 29 Jul 2026 19:31:15 +0800 Subject: [PATCH 1/2] fix: internal trace calls should not compete the semaphore --- src/daemon/mod.rs | 43 +++++++++++++++++++++++++++------- src/rpc/methods/eth.rs | 19 +++++++++------ src/rpc/methods/eth/types.rs | 7 ++++++ src/rpc/methods/state.rs | 6 ++++- src/state_manager/execution.rs | 23 ++++++++++++++---- src/state_manager/tests.rs | 9 +++++-- src/state_manager/utils.rs | 12 ++++++++-- 7 files changed, 95 insertions(+), 24 deletions(-) diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 27a51a09aaf..29b7d33023e 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -25,6 +25,7 @@ use crate::networks::{self, ChainConfig}; use crate::prelude::*; use crate::rpc::RPCState; use crate::rpc::eth::filter::EthEventHandler; +use crate::rpc::eth::types::CallSource; use crate::rpc::start_rpc; use crate::shim::address::Address; use crate::shim::clock::ChainEpoch; @@ -378,6 +379,10 @@ fn maybe_prefill_rpc_caches( // Skip if the node is catching up to avoid unnecessary work, as the head may be changing rapidly. continue; } + Ok(tsk) if state_manager.chain_store().heaviest_tipset().key() != &tsk => { + // Skip if the tipset has already been superseded + continue; + } Ok(tsk) => { let state_manager = state_manager.shallow_clone(); let cancellation_token = cancellation_token.clone(); @@ -386,6 +391,7 @@ fn maybe_prefill_rpc_caches( .run_until_cancelled(prefill_rpc_caches_for_tipset( state_manager, tsk, + cancellation_token.clone(), )) .await }); @@ -400,7 +406,11 @@ fn maybe_prefill_rpc_caches( } } -async fn prefill_rpc_caches_for_tipset(state_manager: StateManager, tsk: TipsetKey) { +async fn prefill_rpc_caches_for_tipset( + state_manager: StateManager, + tsk: TipsetKey, + cancellation_token: CancellationToken, +) { match state_manager.chain_index().load_required_tipset(&tsk) { Ok(ts) => { { @@ -410,6 +420,30 @@ async fn prefill_rpc_caches_for_tipset(state_manager: StateManager, tsk: TipsetK return; // Skip when state computation fails } } + { + // Warms both the FVM-replay cache and the parity-trace cache, + // since `eth_trace_block` calls `execution_trace` internally. + // Note that we do not block the loop here as the trace computation can be expensive. + // Also, we skip this tipset when it has already been superseded + if state_manager.chain_store().heaviest_tipset().key() == ts.key() { + tokio::spawn({ + let state_manager = state_manager.shallow_clone(); + let ts = ts.shallow_clone(); + async move { + if let Some(Err(e)) = cancellation_token + .run_until_cancelled(crate::rpc::eth::eth_trace_block( + &state_manager, + &ts, + CallSource::Internal, + )) + .await + { + warn!("failed to call `eth_trace_block` for cache warmup: {e:#}"); + } + } + }); + } + } for tx_info in [crate::rpc::eth::TxInfo::Full, crate::rpc::eth::TxInfo::Hash] { if let Err(e) = crate::rpc::eth::Block::from_filecoin_tipset( &state_manager, @@ -421,13 +455,6 @@ async fn prefill_rpc_caches_for_tipset(state_manager: StateManager, tsk: TipsetK warn!("failed to call `Block::from_filecoin_tipset` for cache warmup: {e:#}"); } } - { - // Warms both the FVM-replay cache and the parity-trace cache, - // since `eth_trace_block` calls `execution_trace` internally. - if let Err(e) = crate::rpc::eth::eth_trace_block(&state_manager, &ts).await { - warn!("failed to call `eth_trace_block` for cache warmup: {e:#}"); - } - } { use crate::rpc::eth::filter::{Matcher, SkipEvent}; struct CollectEventsCachePrefillingMatcher; diff --git a/src/rpc/methods/eth.rs b/src/rpc/methods/eth.rs index 499876d541b..3c33ee6cb1f 100644 --- a/src/rpc/methods/eth.rs +++ b/src/rpc/methods/eth.rs @@ -3538,7 +3538,7 @@ impl RpcMethod<1> for EthTraceBlock { let ts = resolver .tipset_by_block_number_or_hash(block_param, ResolveNullTipset::Fail) .await?; - eth_trace_block(&ctx.state_manager, &ts) + eth_trace_block(&ctx.state_manager, &ts, CallSource::External) .await .map(NotNullVec) } @@ -3548,8 +3548,9 @@ impl RpcMethod<1> for EthTraceBlock { async fn execute_tipset_traces( state_manager: &StateManager, ts: &Tipset, + source: CallSource, ) -> Result<(StateTree, Vec), ServerError> { - let (state_root, raw_traces) = state_manager.execution_trace(ts).await?; + let (state_root, raw_traces) = state_manager.execution_trace(ts, source).await?; let state = state_manager.get_state_tree(&state_root)?; // Resolve every non-system message's tx hash in parallel. Each lookup is @@ -3604,6 +3605,7 @@ fn non_system_traces_with_positions( pub(crate) async fn eth_trace_block( state_manager: &StateManager, ts: &Tipset, + source: CallSource, ) -> Result, ServerError> { // 64 most-recent blocks; bounded by count, not bytes (a few MiB on mainnet, // see the `cache_eth_trace_block_size` metric). @@ -3616,7 +3618,7 @@ pub(crate) async fn eth_trace_block( let block_cid = ts.key().cid()?; let traces = ETH_TRACE_BLOCK_CACHE .get_or_insert_async(&CidWrapper::from(block_cid), async move { - let (state, entries) = execute_tipset_traces(state_manager, ts).await?; + let (state, entries) = execute_tipset_traces(state_manager, ts, source).await?; let block_hash: EthHash = block_cid.into(); let mut all_traces = vec![]; @@ -3665,6 +3667,7 @@ impl RpcMethod<2> for EthDebugTraceTransaction { tx_hash, opts, &cancellation_token, + CallSource::External, ) .await } @@ -3676,6 +3679,7 @@ async fn debug_trace_transaction( tx_hash: String, opts: GethDebugTracingOptions, cancellation_token: &CancellationToken, + source: CallSource, ) -> Result { let tracer = match &opts.tracer { Some(t) => t.clone(), @@ -3752,7 +3756,7 @@ async fn debug_trace_transaction( return Ok(GethTrace::PreState(frame)); } - let (state, entries) = execute_tipset_traces(&ctx.state_manager, &ts).await?; + let (state, entries) = execute_tipset_traces(&ctx.state_manager, &ts, source).await?; let entry = entries .into_iter() .find(|e| e.tx_hash == eth_hash) @@ -3956,7 +3960,7 @@ impl RpcMethod<1> for EthTraceTransaction { .tipset_by_block_number_or_hash(eth_txn.block_number, ResolveNullTipset::TakeOlder) .await?; - let traces = eth_trace_block(&ctx.state_manager, &ts) + let traces = eth_trace_block(&ctx.state_manager, &ts, CallSource::External) .await? .into_iter() .filter(|trace| trace.transaction_hash == eth_hash) @@ -3996,7 +4000,7 @@ impl RpcMethod<2> for EthTraceReplayBlockTransactions { .tipset_by_block_number_or_hash(block_param, ResolveNullTipset::Fail) .await?; - eth_trace_replay_block_transactions(&ctx, &ts) + eth_trace_replay_block_transactions(&ctx, &ts, CallSource::External) .await .map(NotNullVec) } @@ -4005,8 +4009,9 @@ impl RpcMethod<2> for EthTraceReplayBlockTransactions { async fn eth_trace_replay_block_transactions( ctx: &Ctx, ts: &Tipset, + source: CallSource, ) -> Result, ServerError> { - let (state, entries) = execute_tipset_traces(&ctx.state_manager, ts).await?; + let (state, entries) = execute_tipset_traces(&ctx.state_manager, ts, source).await?; let mut all_traces = vec![]; for entry in entries { diff --git a/src/rpc/methods/eth/types.rs b/src/rpc/methods/eth/types.rs index e39e617f4d4..be0bec6fbfa 100644 --- a/src/rpc/methods/eth/types.rs +++ b/src/rpc/methods/eth/types.rs @@ -17,6 +17,13 @@ pub const METHOD_GET_STORAGE_AT: u64 = 5; const UNCOMPRESSED_PUBLIC_KEY_SIZE: usize = 65; +/// Source of a method call +#[derive(Debug, Copy, Clone)] +pub enum CallSource { + Internal, + External, +} + #[derive( Eq, Hash, diff --git a/src/rpc/methods/state.rs b/src/rpc/methods/state.rs index 44f912c088f..4aa1ad8863c 100644 --- a/src/rpc/methods/state.rs +++ b/src/rpc/methods/state.rs @@ -17,6 +17,7 @@ use crate::libp2p::NetworkMessage; use crate::lotus_json::{LotusJson, lotus_json_with_self}; use crate::networks::{ChainConfig, NetworkChain}; use crate::prelude::*; +use crate::rpc::eth::types::CallSource; use crate::rpc::registry::actors_reg::load_and_serialize_actor_state; use crate::shim::actors::market::DealState; use crate::shim::actors::market::ext::MarketStateExt as _; @@ -157,7 +158,10 @@ impl RpcMethod<2> for StateReplay { _: &http::Extensions, ) -> Result { let tipset = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?; - Ok(ctx.state_manager.replay(tipset, message_cid).await?) + Ok(ctx + .state_manager + .replay(tipset, message_cid, CallSource::External) + .await?) } } diff --git a/src/state_manager/execution.rs b/src/state_manager/execution.rs index 466891c1eae..b4aa6bfa8ed 100644 --- a/src/state_manager/execution.rs +++ b/src/state_manager/execution.rs @@ -7,6 +7,7 @@ use super::state_computation::{ use super::utils::structured; use super::*; use crate::interpreter::{CalledAt, VMTrace}; +use crate::rpc::eth::types::CallSource; use crate::rpc::state::{ApiInvocResult, MessageGasCost}; use anyhow::{Context as _, bail}; use num_traits::identities::Zero; @@ -21,9 +22,14 @@ impl StateManager { /// Lotus, which halts at the target message, this executes the whole /// tipset — the coalescing depends on it, don't port the halt back. /// Consequently, failures after the target message also fail the replay. - pub async fn replay(&self, ts: Tipset, mcid: Cid) -> Result { + pub async fn replay( + &self, + ts: Tipset, + mcid: Cid, + source: CallSource, + ) -> Result { let (_, trace) = self - .execution_trace(&ts) + .execution_trace(&ts, source) .await .map_err(|e| Error::Other(format!("unexpected error during execution : {e}")))?; trace @@ -202,11 +208,15 @@ impl StateManager { pub async fn execution_trace( &self, tipset: &Tipset, + source: CallSource, ) -> anyhow::Result<(Cid, Vec>)> { let key = tipset.key(); let (state_root, invoc_trace) = self .trace_cache - .get_or_insert_async(key, self.execution_trace_inner(tipset.shallow_clone())) + .get_or_insert_async( + key, + self.execution_trace_inner(tipset.shallow_clone(), source), + ) .await?; Ok((state_root.into(), invoc_trace)) } @@ -214,8 +224,13 @@ impl StateManager { async fn execution_trace_inner( &self, tipset: Tipset, + source: CallSource, ) -> anyhow::Result<(CidWrapper, Vec>)> { - let permit = self.replay_permit().await; + // Internal calls like cache prefilling should not compete the semaphore + let permit = match source { + CallSource::External => Some(self.replay_permit().await), + CallSource::Internal => None, + }; let this = self.shallow_clone(); tokio::task::spawn_blocking(move || { let _permit = permit; diff --git a/src/state_manager/tests.rs b/src/state_manager/tests.rs index 1f367e32011..835a097bcf0 100644 --- a/src/state_manager/tests.rs +++ b/src/state_manager/tests.rs @@ -3,6 +3,7 @@ use super::*; use crate::db::MemoryDB; +use crate::rpc::eth::types::CallSource; use crate::shim::executor::StampedEvent; use fil_actors_shared::fvm_ipld_amt::Amt; @@ -343,7 +344,7 @@ async fn replay_is_served_from_the_tipset_trace_cache() { (Cid::default().into(), vec![Arc::new(cached.clone())]), ); - let replayed = sm.replay(ts, mcid).await.unwrap(); + let replayed = sm.replay(ts, mcid, CallSource::External).await.unwrap(); assert_eq!(replayed, cached); } @@ -366,7 +367,11 @@ async fn replay_of_message_absent_from_cached_trace_fails_without_executing() { .insert(ts.key().clone(), (Cid::default().into(), vec![])); let err = sm - .replay(ts, Cid::from_cbor_blake2b256(&"absent-message").unwrap()) + .replay( + ts, + Cid::from_cbor_blake2b256(&"absent-message").unwrap(), + CallSource::External, + ) .await .unwrap_err(); // "failed to replay" is the message-not-found contract exposed via RPC. diff --git a/src/state_manager/utils.rs b/src/state_manager/utils.rs index 949564a01a0..d8676bf6b2b 100644 --- a/src/state_manager/utils.rs +++ b/src/state_manager/utils.rs @@ -359,6 +359,8 @@ pub mod state_compute { use super::*; #[cfg(feature = "cargo-test")] use crate::chain_sync::tipset_syncer::validate_tipset; + #[cfg(feature = "cargo-test")] + use crate::rpc::eth::types::CallSource; #[tokio::test(flavor = "multi_thread")] async fn test_list_state_snapshot_files() { @@ -402,7 +404,10 @@ pub mod state_compute { .expect("test tipset must contain messages") .cid(); - let replayed = sm.replay(ts.clone(), msg_cid).await.unwrap(); + let replayed = sm + .replay(ts.clone(), msg_cid, CallSource::External) + .await + .unwrap(); assert_eq!(replayed.msg_cid, msg_cid); let (_, trace) = sm @@ -417,7 +422,10 @@ pub mod state_compute { // A second replay of the same tipset must not re-execute it. let misses = sm.trace_cache.misses(); - let replayed_again = sm.replay(ts.clone(), msg_cid).await.unwrap(); + let replayed_again = sm + .replay(ts.clone(), msg_cid, CallSource::External) + .await + .unwrap(); assert_eq!(replayed_again, replayed); assert_eq!(sm.trace_cache.misses(), misses); } From dbbf184923d099f96ba1ff3e619cff39d7a42ca3 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Wed, 29 Jul 2026 21:56:39 +0800 Subject: [PATCH 2/2] cover both CallSources in tests --- src/state_manager/tests.rs | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/src/state_manager/tests.rs b/src/state_manager/tests.rs index 835a097bcf0..07845b460e8 100644 --- a/src/state_manager/tests.rs +++ b/src/state_manager/tests.rs @@ -6,6 +6,7 @@ use crate::db::MemoryDB; use crate::rpc::eth::types::CallSource; use crate::shim::executor::StampedEvent; use fil_actors_shared::fvm_ipld_amt::Amt; +use rstest::rstest; fn create_raw_event_v4(emitter: u64, key: &str) -> fvm_shared4::event::StampedEvent { fvm_shared4::event::StampedEvent { @@ -329,8 +330,10 @@ fn state_manager_with_unexecutable_tipset() -> (StateManager, Tipset) { (sm, ts) } -#[tokio::test] -async fn replay_is_served_from_the_tipset_trace_cache() { +#[rstest] +#[case(CallSource::External)] +#[case(CallSource::Internal)] +fn replay_is_served_from_the_tipset_trace_cache(#[case] source: CallSource) { use crate::utils::cid::CidCborExt; let (sm, ts) = state_manager_with_unexecutable_tipset(); @@ -344,7 +347,7 @@ async fn replay_is_served_from_the_tipset_trace_cache() { (Cid::default().into(), vec![Arc::new(cached.clone())]), ); - let replayed = sm.replay(ts, mcid, CallSource::External).await.unwrap(); + let replayed = tokio_test::block_on(sm.replay(ts, mcid, source)).unwrap(); assert_eq!(replayed, cached); } @@ -358,22 +361,22 @@ async fn replay_permits_are_sized_by_configured_concurrency() { assert_eq!(sm.replay_semaphore.available_permits(), permits - 1); } -#[tokio::test] -async fn replay_of_message_absent_from_cached_trace_fails_without_executing() { +#[rstest] +#[case(CallSource::External)] +#[case(CallSource::Internal)] +fn replay_of_message_absent_from_cached_trace_fails_without_executing(#[case] source: CallSource) { use crate::utils::cid::CidCborExt; let (sm, ts) = state_manager_with_unexecutable_tipset(); sm.trace_cache .insert(ts.key().clone(), (Cid::default().into(), vec![])); - let err = sm - .replay( - ts, - Cid::from_cbor_blake2b256(&"absent-message").unwrap(), - CallSource::External, - ) - .await - .unwrap_err(); + let err = tokio_test::block_on(sm.replay( + ts, + Cid::from_cbor_blake2b256(&"absent-message").unwrap(), + source, + )) + .unwrap_err(); // "failed to replay" is the message-not-found contract exposed via RPC. assert!( matches!(err, Error::Other(ref s) if s == "failed to replay"),