perf(trie): parallelize HashedPostStateSorted::from_reverts

Parallelizes the hashing/sorting step using Rayon when account count
exceeds a threshold (10000). This alleviates CPU bottlenecks during large
state reverts or deep reorgs.

Closes #20049

Cargo.lock

@@ -10919,6 +10919,7 @@ dependencies = [
  "alloy-rlp",
  "proptest",
  "proptest-arbitrary-interop",
+ "rayon",
  "reth-chainspec",
  "reth-db",
  "reth-db-api",

crates/trie/db/Cargo.toml

@@ -24,6 +24,9 @@ alloy-primitives.workspace = true
 # tracing
 tracing.workspace = true
 
+# rayon
+rayon = { workspace = true, optional = true }
+
 [dev-dependencies]
 # reth
 reth-chainspec.workspace = true
@@ -48,6 +51,21 @@ serde_json.workspace = true
 similar-asserts.workspace = true
 
 [features]
+default = ["std"]
+std = [
+    "dep:rayon",
+    "alloy-consensus/std",
+    "alloy-primitives/std",
+    "alloy-rlp/std",
+    "reth-chainspec/std",
+    "reth-execution-errors/std",
+    "reth-primitives-traits/std",
+    "revm/std",
+    "revm-database/std",
+    "serde_json/std",
+    "reth-trie-common/std",
+    "tracing/std",
+]
 metrics = ["reth-trie/metrics"]
 serde = [
     "similar-asserts/serde",

crates/trie/db/src/state.rs

@@ -19,6 +19,9 @@ use std::{
 };
 use tracing::{debug, instrument};
 
+#[cfg(feature = "std")]
+use rayon::iter::{IntoParallelIterator, ParallelIterator};
+
 /// Extends [`StateRoot`] with operations specific for working with a database transaction.
 pub trait DatabaseStateRoot<'a, TX>: Sized {
     /// Create a new [`StateRoot`] instance.
@@ -280,14 +283,39 @@ impl<TX: DbTx> DatabaseHashedPostState<TX> for HashedPostStateSorted {
             }
         }
 
-        // Sort storage slots and convert to HashedStorageSorted
-        let hashed_storages = storages
-            .into_iter()
-            .map(|(address, mut slots)| {
-                slots.sort_unstable_by_key(|(slot, _)| *slot);
-                (address, HashedStorageSorted { storage_slots: slots, wiped: false })
-            })
-            .collect();
+        // Threshold based on benchmark
+        const PARALLEL_THRESHOLD: usize = 10_000;
+
+        #[cfg(feature = "std")]
+        let use_parallel = storages.len() >= PARALLEL_THRESHOLD;
+
+        #[cfg(not(feature = "std"))]
+        let use_parallel = false;
+
+        let hashed_storages = if use_parallel {
+            #[cfg(feature = "std")]
+            {
+                storages
+                    .into_par_iter()
+                    .map(|(address, mut slots)| {
+                        slots.sort_unstable_by_key(|(slot, _)| *slot);
+                        (address, HashedStorageSorted { storage_slots: slots, wiped: false })
+                    })
+                    .collect()
+            }
+            #[cfg(not(feature = "std"))]
+            {
+                unreachable!()
+            }
+        } else {
+            storages
+                .into_iter()
+                .map(|(address, mut slots)| {
+                    slots.sort_unstable_by_key(|(slot, _)| *slot);
+                    (address, HashedStorageSorted { storage_slots: slots, wiped: false })
+                })
+                .collect()
+        };
 
         Ok(Self::new(accounts, hashed_storages))
     }