perf: parallelize attestation signature verification with rayon

Signature verification dominated block processing time (~40-50ms per
attestation, sequential). With 36 attestations this meant ~1.4-1.9s per
block, causing the target to fall hundreds of slots behind the head.

Split into two phases:
1. Prepare inputs sequentially (cheap: bit checks, pubkey lookups)
2. Verify signatures in parallel via rayon's par_iter

On a 4-core machine with 36 attestations, expect ~4x speedup
(1.4s -> ~350ms).

Author: MegaRedHand

Cargo.lock

@@ -66,6 +66,7 @@ libssz-types = "0.2"
 # Build-time version info
 vergen-git2 = { version = "9", features = ["rustc"] }
 
+rayon = "1.10"
 rand = "0.9"
 rocksdb = "0.24"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }

Cargo.toml

@@ -23,6 +23,7 @@ spawned-concurrency.workspace = true
 
 tokio.workspace = true
 
+rayon.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 

crates/blockchain/Cargo.toml

@@ -1164,40 +1164,55 @@ fn verify_signatures(
     let validators = &state.validators;
     let num_validators = validators.len() as u64;
 
-    // Verify each attestation's signature proof
+    // Verify each attestation's signature proof in parallel
     let aggregated_start = std::time::Instant::now();
-    for (attestation, aggregated_proof) in attestations.iter().zip(attestation_signatures) {
-        if attestation.aggregation_bits != aggregated_proof.participants {
-            return Err(StoreError::ParticipantsMismatch);
-        }
 
-        let slot: u32 = attestation.data.slot.try_into().expect("slot exceeds u32");
-        let message = attestation.data.hash_tree_root();
+    // Prepare verification inputs sequentially (cheap: bit checks + pubkey lookups)
+    let verification_inputs: Vec<_> = attestations
+        .iter()
+        .zip(attestation_signatures)
+        .map(|(attestation, aggregated_proof)| {
+            if attestation.aggregation_bits != aggregated_proof.participants {
+                return Err(StoreError::ParticipantsMismatch);
+            }
 
-        // Collect public keys with bounds check in a single pass
-        let public_keys: Vec<_> = validator_indices(&attestation.aggregation_bits)
-            .map(|vid| {
-                if vid >= num_validators {
-                    return Err(StoreError::InvalidValidatorIndex);
+            let slot: u32 = attestation.data.slot.try_into().expect("slot exceeds u32");
+            let message = attestation.data.hash_tree_root();
+
+            let public_keys: Vec<_> = validator_indices(&attestation.aggregation_bits)
+                .map(|vid| {
+                    if vid >= num_validators {
+                        return Err(StoreError::InvalidValidatorIndex);
+                    }
+                    validators[vid as usize]
+                        .get_pubkey()
+                        .map_err(|_| StoreError::PubkeyDecodingFailed(vid))
+                })
+                .collect::<Result<_, _>>()?;
+
+            Ok((&aggregated_proof.proof_data, public_keys, message, slot))
+        })
+        .collect::<Result<_, StoreError>>()?;
+
+    // Run expensive signature verification in parallel
+    use rayon::prelude::*;
+    verification_inputs
+        .par_iter()
+        .try_for_each(|(proof_data, public_keys, message, slot)| {
+            let result =
+                verify_aggregated_signature(proof_data, public_keys.clone(), message, *slot);
+            match result {
+                Ok(()) => {
+                    metrics::inc_pq_sig_aggregated_signatures_valid();
+                    Ok(())
+                }
+                Err(e) => {
+                    metrics::inc_pq_sig_aggregated_signatures_invalid();
+                    Err(StoreError::AggregateVerificationFailed(e))
                 }
-                validators[vid as usize]
-                    .get_pubkey()
-                    .map_err(|_| StoreError::PubkeyDecodingFailed(vid))
-            })
-            .collect::<Result<_, _>>()?;
-
-        let verification_result = {
-            let _timing = metrics::time_pq_sig_aggregated_signatures_verification();
-            verify_aggregated_signature(&aggregated_proof.proof_data, public_keys, &message, slot)
-        };
-        match verification_result {
-            Ok(()) => metrics::inc_pq_sig_aggregated_signatures_valid(),
-            Err(e) => {
-                metrics::inc_pq_sig_aggregated_signatures_invalid();
-                return Err(StoreError::AggregateVerificationFailed(e));
             }
-        }
-    }
+        })?;
+
     let aggregated_elapsed = aggregated_start.elapsed();
 
     let proposer_start = std::time::Instant::now();