perf(tree): streaming state root from proofs

crates/engine/service/src/service.rs

@@ -91,6 +91,7 @@ where
             blockchain_db,
             executor_factory,
             consensus,
+            pipeline_task_spawner.clone(),
             payload_validator,
             persistence_handle,
             payload_builder,

crates/engine/tree/Cargo.toml

@@ -20,6 +20,7 @@ reth-consensus.workspace = true
 reth-engine-primitives.workspace = true
 reth-errors.workspace = true
 reth-evm.workspace = true
+reth-execution-errors.workspace = true
 reth-network-p2p.workspace = true
 reth-payload-builder.workspace = true
 reth-payload-primitives.workspace = true
@@ -33,11 +34,13 @@ reth-stages-api.workspace = true
 reth-tasks.workspace = true
 reth-node-types.workspace = true
 reth-trie.workspace = true
-reth-trie-parallel.workspace = true
+reth-trie-parallel = { workspace = true, features = ["async"] }
+alloy-rlp.workspace = true
 
 # common
 futures.workspace = true
 tokio = { workspace = true, features = ["macros", "sync"] }
+tokio-stream.workspace = true
 thiserror.workspace = true
 
 # metrics

crates/engine/tree/src/tree/metrics.rs

@@ -40,12 +40,15 @@ pub(crate) struct BlockValidationMetrics {
     pub(crate) state_root_histogram: Histogram,
     /// Latest state root duration
     pub(crate) state_root_duration: Gauge,
+    /// Histogram of state root duration
+    pub(crate) state_root_from_proofs_histogram: Histogram,
 }
 
 impl BlockValidationMetrics {
     /// Records a new state root time, updating both the histogram and state root gauge
-    pub(crate) fn record_state_root(&self, elapsed_as_secs: f64) {
+    pub(crate) fn record_state_root(&self, elapsed_as_secs: f64, elapsed_from_proofs_as_secs: f64) {
         self.state_root_duration.set(elapsed_as_secs);
         self.state_root_histogram.record(elapsed_as_secs);
+        self.state_root_from_proofs_histogram.record(elapsed_from_proofs_as_secs);
     }
 }