do not wait for all proofs to start processing accounts trie

crates/trie/parallel/src/proof.rs

@@ -131,8 +131,6 @@ where
             "Starting parallel proof generation"
         );
 
-        let (tx, rx) = std::sync::mpsc::sync_channel(num_threads);
-
         let mut storage_proofs =
             B256HashMap::with_capacity_and_hasher(storage_root_targets.len(), Default::default());
 
@@ -144,7 +142,8 @@ where
             let trie_nodes_sorted = self.nodes_sorted.clone();
             let hashed_state_sorted = self.state_sorted.clone();
             let collect_masks = self.collect_branch_node_hash_masks;
-            let tx = tx.clone();
+
+            let (tx, rx) = std::sync::mpsc::sync_channel(1);
 
             pool.spawn_fifo(move || {
                 debug!(
@@ -181,15 +180,15 @@ where
                     );
 
                     let proof_start = Instant::now();
-                    let proof = StorageProof::new_hashed(
+                    let proof_result = StorageProof::new_hashed(
                         trie_cursor_factory,
                         hashed_cursor_factory,
                         hashed_address,
                     )
                     .with_prefix_set_mut(PrefixSetMut::from(prefix_set.iter().cloned()))
                     .with_branch_node_hash_masks(collect_masks)
                     .storage_multiproof(target_slots)
-                    .map_err(|e| ParallelStateRootError::Other(e.to_string()))?;
+                    .map_err(|e| ParallelStateRootError::Other(e.to_string()));
 
                     trace!(
                         target: "trie::parallel",
@@ -198,7 +197,7 @@ where
                         "Completed proof calculation"
                     );
 
-                    Ok((hashed_address, proof))
+                    proof_result
                 })();
 
                 let task_time = task_start.elapsed();
@@ -212,35 +211,7 @@ where
                     );
                 }
             });
-        }
-
-        // Wait for all proofs
-        for _ in 0..storage_root_targets_len {
-            match rx.recv_timeout(std::time::Duration::from_secs(30)) {
-                Ok(result) => match result {
-                    Ok((address, proof)) => {
-                        storage_proofs.insert(address, proof);
-                    }
-                    Err(e) => {
-                        error!(
-                            target: "trie::parallel",
-                            error = ?e,
-                            "Proof calculation failed"
-                        );
-                        return Err(e);
-                    }
-                },
-                Err(e) => {
-                    error!(
-                        target: "trie::parallel",
-                        error = ?e,
-                        "Failed to receive proof result"
-                    );
-                    return Err(ParallelStateRootError::Other(format!(
-                        "Failed to receive proof result: {e:?}",
-                    )));
-                }
-            }
+            storage_proofs.insert(hashed_address, rx);
         }
 
         let provider_ro = self.view.provider_ro()?;
@@ -284,7 +255,13 @@ where
                 }
                 TrieElement::Leaf(hashed_address, account) => {
                     let storage_multiproof = match storage_proofs.remove(&hashed_address) {
-                        Some(proof) => proof,
+                        Some(rx) => rx.recv().map_err(|_| {
+                            ParallelStateRootError::StorageRoot(StorageRootError::Database(
+                                DatabaseError::Other(format!(
+                                    "channel closed for {hashed_address}"
+                                )),
+                            ))
+                        })??,
                         // Since we do not store all intermediate nodes in the database, there might
                         // be a possibility of re-adding a non-modified leaf to the hash builder.
                         None => {