fix: internal trace calls should not compete the semaphore (#7421)

Author: hanabi1224

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;

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<DbImpl>, Vec<trace::TipsetTraceEntry>), 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<Vec<EthBlockTrace>, 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<GethTrace, ServerError> {
     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<Vec<EthReplayBlockTransactionTrace>, 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 {

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,

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<Self::Ok, ServerError> {
         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?)
     }
 }
 

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<ApiInvocResult, Error> {
+    pub async fn replay(
+        &self,
+        ts: Tipset,
+        mcid: Cid,
+        source: CallSource,
+    ) -> Result<ApiInvocResult, Error> {
         let (_, trace) = self
-            .execution_trace(&ts)
+            .execution_trace(&ts, source)
             .await
             .map_err(|e| Error::Other(format!("unexpected error during execution : {e}")))?;
         trace
@@ -202,20 +208,29 @@ impl StateManager {
     pub async fn execution_trace(
         &self,
         tipset: &Tipset,
+        source: CallSource,
     ) -> anyhow::Result<(Cid, Vec<Arc<ApiInvocResult>>)> {
         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))
     }
 
     async fn execution_trace_inner(
         &self,
         tipset: Tipset,
+        source: CallSource,
     ) -> anyhow::Result<(CidWrapper, Vec<Arc<ApiInvocResult>>)> {
-        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;

src/state_manager/tests.rs

@@ -3,8 +3,10 @@
 
 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;
+use rstest::rstest;
 
 fn create_raw_event_v4(emitter: u64, key: &str) -> fvm_shared4::event::StampedEvent {
     fvm_shared4::event::StampedEvent {
@@ -328,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();
@@ -343,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).await.unwrap();
+    let replayed = tokio_test::block_on(sm.replay(ts, mcid, source)).unwrap();
     assert_eq!(replayed, cached);
 }
 
@@ -357,18 +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())
-        .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"),

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);
         }