fix: internal trace calls should not compete the semaphore

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