perf(tree): worker pooling for storage in multiproof generation

crates/engine/primitives/src/config.rs

@@ -165,6 +165,7 @@ impl TreeConfig {
         prewarm_max_concurrency: usize,
         allow_unwind_canonical_header: bool,
     ) -> Self {
+        assert!(max_proof_task_concurrency > 0, "max_proof_task_concurrency must be at least 1");
         Self {
             persistence_threshold,
             memory_block_buffer_target,
@@ -394,6 +395,7 @@ impl TreeConfig {
         mut self,
         max_proof_task_concurrency: u64,
     ) -> Self {
+        assert!(max_proof_task_concurrency > 0, "max_proof_task_concurrency must be at least 1");
         self.max_proof_task_concurrency = max_proof_task_concurrency;
         self
     }

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

@@ -196,11 +196,17 @@ where
             state_root_config.prefix_sets.clone(),
         );
         let max_proof_task_concurrency = config.max_proof_task_concurrency() as usize;
+        // Default to half of max concurrency, leaving room for on-demand tasks (Accountproof and
+        // blinded nodes)
+        let storage_worker_count = (max_proof_task_concurrency / 2)
+            .max(1)
+            .min(max_proof_task_concurrency.saturating_sub(1));
         let proof_task = ProofTaskManager::new(
             self.executor.handle().clone(),
             state_root_config.consistent_view.clone(),
             task_ctx,
             max_proof_task_concurrency,
+            storage_worker_count,
         );
 
         // We set it to half of the proof task concurrency, because often for each multiproof we

crates/engine/tree/src/tree/payload_processor/multiproof.rs

@@ -1236,6 +1236,7 @@ mod tests {
             config.consistent_view.clone(),
             task_ctx,
             1,
+            1, // storage_worker_count: 1 for tests
         );
         let channel = channel();
 

crates/trie/parallel/Cargo.toml

@@ -36,6 +36,7 @@ derive_more.workspace = true
 rayon.workspace = true
 itertools.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread"] }
+crossbeam-channel.workspace = true
 
 # `metrics` feature
 reth-metrics = { workspace = true, optional = true }

crates/trie/parallel/src/proof.rs

@@ -448,7 +448,7 @@ mod tests {
         let task_ctx =
             ProofTaskCtx::new(Default::default(), Default::default(), Default::default());
         let proof_task =
-            ProofTaskManager::new(rt.handle().clone(), consistent_view.clone(), task_ctx, 1);
+            ProofTaskManager::new(rt.handle().clone(), consistent_view.clone(), task_ctx, 1, 1);
         let proof_task_handle = proof_task.handle();
 
         // keep the join handle around to make sure it does not return any errors