perf(tree): state provider builder

crates/engine/tree/src/tree/mod.rs

@@ -439,6 +439,43 @@ impl<N: NodePrimitives> TreeState<N> {
     const fn canonical_block_number(&self) -> BlockNumber {
         self.canonical_head().number
     }
+
+    /// Returns a builder for creating state providers for the given hash.
+    ///
+    /// This is used to optimize state provider creation in parallel execution contexts.
+    pub fn state_provider_builder<P>(
+        &self,
+        provider: P,
+        hash: B256,
+    ) -> Option<StateProviderBuilder<N, P>>
+    where
+        P: BlockReader + StateProviderFactory + StateReader + StateCommitmentProvider + Clone,
+    {
+        let (historical, blocks) = self.blocks_by_hash(hash)?;
+        Some(StateProviderBuilder { provider, historical, blocks })
+    }
+}
+
+/// A builder for creating state providers that can be used across threads.
+#[derive(Clone, Debug)]
+pub struct StateProviderBuilder<N: NodePrimitives, P> {
+    /// The provider instance used to fetch historical state.
+    provider: P,
+    /// The historical block hash to fetch state from.
+    historical: B256,
+    /// The blocks that form the chain from historical to target.
+    blocks: Vec<ExecutedBlockWithTrieUpdates<N>>,
+}
+
+impl<N: NodePrimitives, P> StateProviderBuilder<N, P>
+where
+    P: BlockReader + StateProviderFactory + StateReader + StateCommitmentProvider + Clone,
+{
+    /// Creates a new state provider from this builder.
+    pub fn build(&self) -> ProviderResult<StateProviderBox> {
+        let historical = self.provider.state_by_block_hash(self.historical)?;
+        Ok(Box::new(MemoryOverlayStateProvider::new(historical, self.blocks.clone())))
+    }
 }
 
 /// Tracks the state of the engine api internals.
@@ -2642,20 +2679,33 @@ where
         state_root_sender: Option<Sender<StateRootMessage>>,
         cancel_execution: ManualCancel,
     ) -> Result<(), InsertBlockErrorKind> {
-        let Some(state_provider) = self.state_provider(block.parent_hash())? else {
-            trace!(target: "engine::tree", parent=%block.parent_hash(), "Could not get state provider for prewarm");
+        // Get the builder once, outside the thread
+        let Some(state_builder) = self.state_provider_builder(block.parent_hash())? else {
+            trace!(target: "engine::tree", parent=%block.parent_hash(), "Could not get state provider builder for prewarm");
             return Ok(())
         };
 
-        // Use the caches to create a new executor
-        let state_provider =
-            CachedStateProvider::new_with_caches(state_provider, caches, cache_metrics);
-
         // clone and copy info required for execution
         let evm_config = self.evm_config.clone();
 
         // spawn task executing the individual tx
         self.thread_pool.spawn(move || {
+            // Create the state provider inside the thread
+            let state_provider = match state_builder.build() {
+                Ok(provider) => provider,
+                Err(err) => {
+                    trace!(target: "engine::tree", %err, "Failed to build state provider in prewarm thread");
+                    return
+                }
+            };
+
+            // Use the caches to create a new provider with caching
+            let state_provider = CachedStateProvider::new_with_caches(
+                state_provider,
+                caches,
+                cache_metrics,
+            );
+
             let state_provider = StateProviderDatabase::new(&state_provider);
 
             // create a new executor and disable nonce checks in the env
@@ -3015,6 +3065,40 @@ where
         );
         Ok(())
     }
+
+    /// Returns a builder for creating state providers for the given hash.
+    ///
+    /// This is an optimization for parallel execution contexts where we want to avoid
+    /// creating state providers in the critical path.
+    pub fn state_provider_builder(
+        &self,
+        hash: B256,
+    ) -> ProviderResult<Option<StateProviderBuilder<N, P>>>
+    where
+        P: BlockReader + StateProviderFactory + StateReader + StateCommitmentProvider + Clone,
+    {
+        // Check if we have the block in memory first
+        if let Some(builder) =
+            self.state.tree_state.state_provider_builder(self.provider.clone(), hash)
+        {
+            return Ok(Some(builder))
+        }
+
+        // Check if the block is persisted
+        if let Some(header) = self.provider.header(&hash)? {
+            debug!(target: "engine::tree", %hash, number = %header.number(), "found canonical state for block in database");
+            // For persisted blocks, we create a builder that will fetch state directly from the
+            // database
+            return Ok(Some(StateProviderBuilder {
+                provider: self.provider.clone(),
+                historical: hash,
+                blocks: vec![],
+            }))
+        }
+
+        debug!(target: "engine::tree", %hash, "no canonical state found for block");
+        Ok(None)
+    }
 }
 
 /// Block inclusion can be valid, accepted, or invalid. Invalid blocks are returned as an error