feat(flashblocks): parallelize sender recovery

  Addresses #282 - reduces sender recovery overhead during flashblock processing.

  ## Changes
  - Add rayon for parallel iteration over transactions
  - Batch ECDSA sender recovery upfront using par_iter()
  - Add sender_recovery_duration metric for observability
  - Preserve cache lookup behavior (check prev_pending_blocks first)
  - Maintain original error handling semantics

  ## Technical Approach
  Follows upstream reth pattern (paradigmxyz/reth#20169) - recover all
  senders in parallel before the sequential transaction processing loop.

  ## Testing
  All 10 existing flashblocks tests pass.

  ## Note
  Benchmarks not included as benchmark infrastructure was removed from main.
  The new sender_recovery_duration metric can be used for production measurement.

Author: madisoncarter1234

Cargo.lock

@@ -140,6 +140,7 @@ lru = "0.16.2"
 rand = "0.9.2"
 uuid = "1.19.0"
 time = "0.3.44"
+rayon = "1.11"
 clap = "4.5.53"
 eyre = "0.6.12"
 bytes = "1.11.0"

Cargo.toml

@@ -54,6 +54,7 @@ tracing.workspace = true
 metrics.workspace = true
 arc-swap.workspace = true
 metrics-derive.workspace = true
+rayon.workspace = true
 
 [dev-dependencies]
 rand.workspace = true

crates/flashblocks/Cargo.toml

@@ -23,6 +23,10 @@ pub struct Metrics {
     #[metric(describe = "Time taken to process a message")]
     pub block_processing_duration: Histogram,
 
+    /// Time spent on parallel sender recovery (ECDSA operations).
+    #[metric(describe = "Time spent on parallel sender recovery")]
+    pub sender_recovery_duration: Histogram,
+
     /// Number of Flashblocks that arrive in an unexpected order.
     #[metric(describe = "Number of Flashblocks that arrive in an unexpected order")]
     pub unexpected_block_order: Counter,

crates/flashblocks/src/metrics.rs

@@ -11,7 +11,7 @@ use alloy_consensus::{
     transaction::{Recovered, SignerRecoverable, TransactionMeta},
 };
 use alloy_eips::BlockNumberOrTag;
-use alloy_primitives::{B256, BlockNumber, Bytes, Sealable, map::foldhash::HashMap};
+use alloy_primitives::{Address, B256, BlockNumber, Bytes, Sealable, map::foldhash::HashMap};
 use alloy_rpc_types::{TransactionTrait, Withdrawal};
 use alloy_rpc_types_engine::{ExecutionPayloadV1, ExecutionPayloadV2, ExecutionPayloadV3};
 use alloy_rpc_types_eth::state::StateOverride;
@@ -21,6 +21,7 @@ use eyre::eyre;
 use op_alloy_consensus::OpTxEnvelope;
 use op_alloy_network::TransactionResponse;
 use op_alloy_rpc_types::Transaction;
+use rayon::prelude::*;
 use reth::{
     chainspec::{ChainSpecProvider, EthChainSpec},
     providers::{BlockReaderIdExt, StateProviderFactory},
@@ -393,16 +394,33 @@ where
             let mut gas_used = 0;
             let mut next_log_index = 0;
 
+            // Parallel sender recovery - batch all ECDSA operations upfront
+            let recovery_start = Instant::now();
+            let sender_by_hash: HashMap<B256, Address> = block
+                .body
+                .transactions
+                .par_iter()
+                .map(|tx| -> eyre::Result<(B256, Address)> {
+                    let tx_hash = tx.tx_hash();
+                    let sender = match prev_pending_blocks
+                        .as_ref()
+                        .and_then(|p| p.get_transaction_sender(&tx_hash))
+                    {
+                        Some(cached) => cached,
+                        None => tx.recover_signer()?,
+                    };
+                    Ok((tx_hash, sender))
+                })
+                .collect::<eyre::Result<_>>()?;
+            self.metrics.sender_recovery_duration.record(recovery_start.elapsed());
+
             for (idx, transaction) in block.body.transactions.iter().enumerate() {
                 let tx_hash = transaction.tx_hash();
 
-                let sender = match prev_pending_blocks
-                    .as_ref()
-                    .and_then(|p| p.get_transaction_sender(&tx_hash))
-                {
-                    Some(sender) => sender,
-                    None => transaction.recover_signer()?,
-                };
+                // Sender already recovered in parallel phase
+                let sender = *sender_by_hash
+                    .get(&tx_hash)
+                    .expect("sender must exist from parallel recovery");
 
                 pending_blocks_builder.with_transaction_sender(tx_hash, sender);
                 pending_blocks_builder.increment_nonce(sender);