perf: parallelize attestation signature verification (#274)

## Summary

- Parallelize attestation signature verification using rayon's
`par_iter`
- Signature verification was sequential (~40-50ms per attestation),
causing blocks with many attestations to take 1.4-1.9s to process
- Split into two phases: prepare inputs sequentially (cheap), then
verify in parallel
- Observed in devnet: with 36 attestations, block processing took
~1.4-1.9s; expect ~4x speedup on 4 cores

## Test plan

- [x] `make fmt` passes
- [x] `make lint` passes
- [x] `make test` passes (all 101 tests, including signature spec tests)
- [ ] Deploy to devnet and verify block processing times drop

Author: MegaRedHand

Cargo.lock

@@ -66,6 +66,7 @@ libssz-types = "0.2"
 # Build-time version info
 vergen-git2 = { version = "9", features = ["rustc"] }
 
+rayon = "1.11"
 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,58 @@ 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.
+    // into_par_iter() moves each tuple, avoiding a clone of public_keys.
+    use rayon::prelude::*;
+    verification_inputs.into_par_iter().try_for_each(
+        |(proof_data, public_keys, message, slot)| {
+            let result = {
+                let _timing = metrics::time_pq_sig_aggregated_signatures_verification();
+                verify_aggregated_signature(proof_data, public_keys, &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();