Batch BLS verification for attestations (#2399)

## Issue Addressed

NA

## Proposed Changes

Adds the ability to verify batches of aggregated/unaggregated attestations from the network.

When the `BeaconProcessor` finds there are messages in the aggregated or unaggregated attestation queues, it will first check the length of the queue:

- `== 1` verify the attestation individually.
- `>= 2` take up to 64 of those attestations and verify them in a batch.

Notably, we only perform batch verification if the queue has a backlog. We don't apply any artificial delays to attestations to try and force them into batches. 

### Batching Details

To assist with implementing batches we modify `beacon_chain::attestation_verification` to have two distinct categories for attestations:

- *Indexed* attestations: those which have passed initial validation and were valid enough for us to derive an `IndexedAttestation`.
- *Verified* attestations: those attestations which were indexed *and also* passed signature verification. These are well-formed, interesting messages which were signed by validators.

The batching functions accept `n` attestations and then return `n` attestation verification `Result`s, where those `Result`s can be any combination of `Ok` or `Err`. In other words, we attempt to verify as many attestations as possible and return specific per-attestation results so peer scores can be updated, if required.

When we batch verify attestations, we first try to map all those attestations to *indexed* attestations. If any of those attestations were able to be indexed, we then perform batch BLS verification on those indexed attestations. If the batch verification succeeds, we convert them into *verified* attestations, disabling individual signature checking. If the batch fails, we convert to verified attestations with individual signature checking enabled.

Ultimately, we optimistically try to do a batch verification of attestation signatures and fall-back to individual verification if it fails. This opens an attach vector for "poisoning" the attestations and causing us to waste a batch verification. I argue that peer scoring should do a good-enough job of defending against this and the typical-case gains massively outweigh the worst-case losses.

## Additional Info

Before this PR, attestation verification took the attestations by value (instead of by reference). It turns out that this was unnecessary and, in my opinion, resulted in some undesirable ergonomics (e.g., we had to pass the attestation back in the `Err` variant to avoid clones). In this PR I've modified attestation verification so that it now takes a reference.

I refactored the `beacon_chain/tests/attestation_verification.rs` tests so they use a builder-esque "tester" struct instead of a weird macro. It made it easier for me to test individual/batch with the same set of tests and I think it was a nice tidy-up. Notably, I did this last to try and make sure my new refactors to *actual* production code would pass under the existing test suite.

Author: paulhauner

beacon_node/beacon_chain/src/attestation_verification.rs

@@ -0,0 +1,222 @@
+//! These two `batch_...` functions provide verification of batches of attestations. They provide
+//! significant CPU-time savings by performing batch verification of BLS signatures.
+//!
+//! In each function, attestations are "indexed" (i.e., the `IndexedAttestation` is computed), to
+//! determine if they should progress to signature verification. Then, all attestations which were
+//! successfully indexed have their signatures verified in a batch. If that signature batch fails
+//! then all attestation signatures are verified independently.
+//!
+//! The outcome of each function is a `Vec<Result>` with a one-to-one mapping to the attestations
+//! supplied as input. Each result provides the exact success or failure result of the corresponding
+//! attestation, with no loss of fidelity when compared to individual verification.
+use super::{
+    CheckAttestationSignature, Error, IndexedAggregatedAttestation, IndexedUnaggregatedAttestation,
+    VerifiedAggregatedAttestation, VerifiedUnaggregatedAttestation,
+};
+use crate::{
+    beacon_chain::VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT, metrics, BeaconChain, BeaconChainError,
+    BeaconChainTypes,
+};
+use bls::verify_signature_sets;
+use state_processing::signature_sets::{
+    indexed_attestation_signature_set_from_pubkeys, signed_aggregate_selection_proof_signature_set,
+    signed_aggregate_signature_set,
+};
+use std::borrow::Cow;
+use types::*;
+
+/// Verify aggregated attestations using batch BLS signature verification.
+///
+/// See module-level docs for more info.
+pub fn batch_verify_aggregated_attestations<'a, T, I>(
+    aggregates: I,
+    chain: &BeaconChain<T>,
+) -> Result<Vec<Result<VerifiedAggregatedAttestation<'a, T>, Error>>, Error>
+where
+    T: BeaconChainTypes,
+    I: Iterator<Item = &'a SignedAggregateAndProof<T::EthSpec>> + ExactSizeIterator,
+{
+    let mut num_indexed = 0;
+    let mut num_failed = 0;
+
+    // Perform indexing of all attestations, collecting the results.
+    let indexing_results = aggregates
+        .map(|aggregate| {
+            let result = IndexedAggregatedAttestation::verify(aggregate, chain);
+            if result.is_ok() {
+                num_indexed += 1;
+            } else {
+                num_failed += 1;
+            }
+            result
+        })
+        .collect::<Vec<_>>();
+
+    // May be set to `No` if batch verification succeeds.
+    let mut check_signatures = CheckAttestationSignature::Yes;
+
+    // Perform batch BLS verification, if any attestation signatures are worth checking.
+    if num_indexed > 0 {
+        let signature_setup_timer =
+            metrics::start_timer(&metrics::ATTESTATION_PROCESSING_BATCH_AGG_SIGNATURE_SETUP_TIMES);
+
+        let pubkey_cache = chain
+            .validator_pubkey_cache
+            .try_read_for(VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT)
+            .ok_or(BeaconChainError::ValidatorPubkeyCacheLockTimeout)?;
+
+        let fork = chain.with_head(|head| Ok::<_, BeaconChainError>(head.beacon_state.fork()))?;
+
+        let mut signature_sets = Vec::with_capacity(num_indexed * 3);
+
+        // Iterate, flattening to get only the `Ok` values.
+        for indexed in indexing_results.iter().flatten() {
+            let signed_aggregate = &indexed.signed_aggregate;
+            let indexed_attestation = &indexed.indexed_attestation;
+
+            signature_sets.push(
+                signed_aggregate_selection_proof_signature_set(
+                    |validator_index| pubkey_cache.get(validator_index).map(Cow::Borrowed),
+                    signed_aggregate,
+                    &fork,
+                    chain.genesis_validators_root,
+                    &chain.spec,
+                )
+                .map_err(BeaconChainError::SignatureSetError)?,
+            );
+            signature_sets.push(
+                signed_aggregate_signature_set(
+                    |validator_index| pubkey_cache.get(validator_index).map(Cow::Borrowed),
+                    signed_aggregate,
+                    &fork,
+                    chain.genesis_validators_root,
+                    &chain.spec,
+                )
+                .map_err(BeaconChainError::SignatureSetError)?,
+            );
+            signature_sets.push(
+                indexed_attestation_signature_set_from_pubkeys(
+                    |validator_index| pubkey_cache.get(validator_index).map(Cow::Borrowed),
+                    &indexed_attestation.signature,
+                    indexed_attestation,
+                    &fork,
+                    chain.genesis_validators_root,
+                    &chain.spec,
+                )
+                .map_err(BeaconChainError::SignatureSetError)?,
+            );
+        }
+
+        metrics::stop_timer(signature_setup_timer);
+
+        let _signature_verification_timer =
+            metrics::start_timer(&metrics::ATTESTATION_PROCESSING_BATCH_AGG_SIGNATURE_TIMES);
+
+        if verify_signature_sets(signature_sets.iter()) {
+            // Since all the signatures verified in a batch, there's no reason for them to be
+            // checked again later.
+            check_signatures = CheckAttestationSignature::No
+        }
+    }
+
+    // Complete the attestation verification, potentially verifying all signatures independently.
+    let final_results = indexing_results
+        .into_iter()
+        .map(|result| match result {
+            Ok(indexed) => {
+                VerifiedAggregatedAttestation::from_indexed(indexed, chain, check_signatures)
+            }
+            Err(e) => Err(e),
+        })
+        .collect();
+
+    Ok(final_results)
+}
+
+/// Verify unaggregated attestations using batch BLS signature verification.
+///
+/// See module-level docs for more info.
+pub fn batch_verify_unaggregated_attestations<'a, T, I>(
+    attestations: I,
+    chain: &BeaconChain<T>,
+) -> Result<Vec<Result<VerifiedUnaggregatedAttestation<'a, T>, Error>>, Error>
+where
+    T: BeaconChainTypes,
+    I: Iterator<Item = (&'a Attestation<T::EthSpec>, Option<SubnetId>)> + ExactSizeIterator,
+{
+    let mut num_partially_verified = 0;
+    let mut num_failed = 0;
+
+    // Perform partial verification of all attestations, collecting the results.
+    let partial_results = attestations
+        .map(|(attn, subnet_opt)| {
+            let result = IndexedUnaggregatedAttestation::verify(attn, subnet_opt, chain);
+            if result.is_ok() {
+                num_partially_verified += 1;
+            } else {
+                num_failed += 1;
+            }
+            result
+        })
+        .collect::<Vec<_>>();
+
+    // May be set to `No` if batch verification succeeds.
+    let mut check_signatures = CheckAttestationSignature::Yes;
+
+    // Perform batch BLS verification, if any attestation signatures are worth checking.
+    if num_partially_verified > 0 {
+        let signature_setup_timer = metrics::start_timer(
+            &metrics::ATTESTATION_PROCESSING_BATCH_UNAGG_SIGNATURE_SETUP_TIMES,
+        );
+
+        let pubkey_cache = chain
+            .validator_pubkey_cache
+            .try_read_for(VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT)
+            .ok_or(BeaconChainError::ValidatorPubkeyCacheLockTimeout)?;
+
+        let fork = chain.with_head(|head| Ok::<_, BeaconChainError>(head.beacon_state.fork()))?;
+
+        let mut signature_sets = Vec::with_capacity(num_partially_verified);
+
+        // Iterate, flattening to get only the `Ok` values.
+        for partially_verified in partial_results.iter().flatten() {
+            let indexed_attestation = &partially_verified.indexed_attestation;
+
+            let signature_set = indexed_attestation_signature_set_from_pubkeys(
+                |validator_index| pubkey_cache.get(validator_index).map(Cow::Borrowed),
+                &indexed_attestation.signature,
+                indexed_attestation,
+                &fork,
+                chain.genesis_validators_root,
+                &chain.spec,
+            )
+            .map_err(BeaconChainError::SignatureSetError)?;
+
+            signature_sets.push(signature_set);
+        }
+
+        metrics::stop_timer(signature_setup_timer);
+
+        let _signature_verification_timer =
+            metrics::start_timer(&metrics::ATTESTATION_PROCESSING_BATCH_UNAGG_SIGNATURE_TIMES);
+
+        if verify_signature_sets(signature_sets.iter()) {
+            // Since all the signatures verified in a batch, there's no reason for them to be
+            // checked again later.
+            check_signatures = CheckAttestationSignature::No
+        }
+    }
+
+    // Complete the attestation verification, potentially verifying all signatures independently.
+    let final_results = partial_results
+        .into_iter()
+        .map(|result| match result {
+            Ok(partial) => {
+                VerifiedUnaggregatedAttestation::from_indexed(partial, chain, check_signatures)
+            }
+            Err(e) => Err(e),
+        })
+        .collect();
+
+    Ok(final_results)
+}

beacon_node/beacon_chain/src/attestation_verification/batch.rs

@@ -1,5 +1,6 @@
 use crate::attestation_verification::{
-    Error as AttestationError, SignatureVerifiedAttestation, VerifiedAggregatedAttestation,
+    batch_verify_aggregated_attestations, batch_verify_unaggregated_attestations,
+    Error as AttestationError, VerifiedAggregatedAttestation, VerifiedAttestation,
     VerifiedUnaggregatedAttestation,
 };
 use crate::attester_cache::{AttesterCache, AttesterCacheKey};
@@ -1510,17 +1511,32 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         })
     }
 
+    /// Performs the same validation as `Self::verify_unaggregated_attestation_for_gossip`, but for
+    /// multiple attestations using batch BLS verification. Batch verification can provide
+    /// significant CPU-time savings compared to individual verification.
+    pub fn batch_verify_unaggregated_attestations_for_gossip<'a, I>(
+        &self,
+        attestations: I,
+    ) -> Result<
+        Vec<Result<VerifiedUnaggregatedAttestation<'a, T>, AttestationError>>,
+        AttestationError,
+    >
+    where
+        I: Iterator<Item = (&'a Attestation<T::EthSpec>, Option<SubnetId>)> + ExactSizeIterator,
+    {
+        batch_verify_unaggregated_attestations(attestations, self)
+    }
+
     /// Accepts some `Attestation` from the network and attempts to verify it, returning `Ok(_)` if
     /// it is valid to be (re)broadcast on the gossip network.
     ///
     /// The attestation must be "unaggregated", that is it must have exactly one
     /// aggregation bit set.
-    pub fn verify_unaggregated_attestation_for_gossip(
+    pub fn verify_unaggregated_attestation_for_gossip<'a>(
         &self,
-        unaggregated_attestation: Attestation<T::EthSpec>,
+        unaggregated_attestation: &'a Attestation<T::EthSpec>,
         subnet_id: Option<SubnetId>,
-    ) -> Result<VerifiedUnaggregatedAttestation<T>, (AttestationError, Attestation<T::EthSpec>)>
-    {
+    ) -> Result<VerifiedUnaggregatedAttestation<'a, T>, AttestationError> {
         metrics::inc_counter(&metrics::UNAGGREGATED_ATTESTATION_PROCESSING_REQUESTS);
         let _timer =
             metrics::start_timer(&metrics::UNAGGREGATED_ATTESTATION_GOSSIP_VERIFICATION_TIMES);
@@ -1539,15 +1555,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         )
     }
 
+    /// Performs the same validation as `Self::verify_aggregated_attestation_for_gossip`, but for
+    /// multiple attestations using batch BLS verification. Batch verification can provide
+    /// significant CPU-time savings compared to individual verification.
+    pub fn batch_verify_aggregated_attestations_for_gossip<'a, I>(
+        &self,
+        aggregates: I,
+    ) -> Result<Vec<Result<VerifiedAggregatedAttestation<'a, T>, AttestationError>>, AttestationError>
+    where
+        I: Iterator<Item = &'a SignedAggregateAndProof<T::EthSpec>> + ExactSizeIterator,
+    {
+        batch_verify_aggregated_attestations(aggregates, self)
+    }
+
     /// Accepts some `SignedAggregateAndProof` from the network and attempts to verify it,
     /// returning `Ok(_)` if it is valid to be (re)broadcast on the gossip network.
-    pub fn verify_aggregated_attestation_for_gossip(
+    pub fn verify_aggregated_attestation_for_gossip<'a>(
         &self,
-        signed_aggregate: SignedAggregateAndProof<T::EthSpec>,
-    ) -> Result<
-        VerifiedAggregatedAttestation<T>,
-        (AttestationError, SignedAggregateAndProof<T::EthSpec>),
-    > {
+        signed_aggregate: &'a SignedAggregateAndProof<T::EthSpec>,
+    ) -> Result<VerifiedAggregatedAttestation<'a, T>, AttestationError> {
         metrics::inc_counter(&metrics::AGGREGATED_ATTESTATION_PROCESSING_REQUESTS);
         let _timer =
             metrics::start_timer(&metrics::AGGREGATED_ATTESTATION_GOSSIP_VERIFICATION_TIMES);
@@ -1597,13 +1623,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// Accepts some attestation-type object and attempts to verify it in the context of fork
     /// choice. If it is valid it is applied to `self.fork_choice`.
     ///
-    /// Common items that implement `SignatureVerifiedAttestation`:
+    /// Common items that implement `VerifiedAttestation`:
     ///
     /// - `VerifiedUnaggregatedAttestation`
     /// - `VerifiedAggregatedAttestation`
     pub fn apply_attestation_to_fork_choice(
         &self,
-        verified: &impl SignatureVerifiedAttestation<T>,
+        verified: &impl VerifiedAttestation<T>,
     ) -> Result<(), Error> {
         let _timer = metrics::start_timer(&metrics::FORK_CHOICE_PROCESS_ATTESTATION_TIMES);
 
@@ -1623,8 +1649,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// and no error is returned.
     pub fn add_to_naive_aggregation_pool(
         &self,
-        unaggregated_attestation: VerifiedUnaggregatedAttestation<T>,
-    ) -> Result<VerifiedUnaggregatedAttestation<T>, AttestationError> {
+        unaggregated_attestation: &impl VerifiedAttestation<T>,
+    ) -> Result<(), AttestationError> {
         let _timer = metrics::start_timer(&metrics::ATTESTATION_PROCESSING_APPLY_TO_AGG_POOL);
 
         let attestation = unaggregated_attestation.attestation();
@@ -1660,7 +1686,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
         };
 
-        Ok(unaggregated_attestation)
+        Ok(())
     }
 
     /// Accepts a `VerifiedSyncCommitteeMessage` and attempts to apply it to the "naive
@@ -1727,13 +1753,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(verified_sync_committee_message)
     }
 
-    /// Accepts a `VerifiedAggregatedAttestation` and attempts to apply it to `self.op_pool`.
+    /// Accepts a `VerifiedAttestation` and attempts to apply it to `self.op_pool`.
     ///
     /// The op pool is used by local block producers to pack blocks with operations.
     pub fn add_to_block_inclusion_pool(
         &self,
-        signed_aggregate: VerifiedAggregatedAttestation<T>,
-    ) -> Result<VerifiedAggregatedAttestation<T>, AttestationError> {
+        verified_attestation: &impl VerifiedAttestation<T>,
+    ) -> Result<(), AttestationError> {
         let _timer = metrics::start_timer(&metrics::ATTESTATION_PROCESSING_APPLY_TO_OP_POOL);
 
         // If there's no eth1 chain then it's impossible to produce blocks and therefore
@@ -1745,15 +1771,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             self.op_pool
                 .insert_attestation(
                     // TODO: address this clone.
-                    signed_aggregate.attestation().clone(),
+                    verified_attestation.attestation().clone(),
                     &fork,
                     self.genesis_validators_root,
                     &self.spec,
                 )
                 .map_err(Error::from)?;
         }
 
-        Ok(signed_aggregate)
+        Ok(())
     }
 
     /// Accepts a `VerifiedSyncContribution` and attempts to apply it to `self.op_pool`.

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -199,6 +199,26 @@ lazy_static! {
         "Time spent on the signature verification of attestation processing"
     );
 
+    /*
+     * Batch Attestation Processing
+     */
+    pub static ref ATTESTATION_PROCESSING_BATCH_AGG_SIGNATURE_SETUP_TIMES: Result<Histogram> = try_create_histogram(
+        "beacon_attestation_processing_batch_agg_signature_setup_times",
+        "Time spent on setting up for the signature verification of batch aggregate processing"
+    );
+    pub static ref ATTESTATION_PROCESSING_BATCH_AGG_SIGNATURE_TIMES: Result<Histogram> = try_create_histogram(
+        "beacon_attestation_processing_batch_agg_signature_times",
+        "Time spent on the signature verification of batch aggregate attestation processing"
+    );
+    pub static ref ATTESTATION_PROCESSING_BATCH_UNAGG_SIGNATURE_SETUP_TIMES: Result<Histogram> = try_create_histogram(
+        "beacon_attestation_processing_batch_unagg_signature_setup_times",
+        "Time spent on setting up for the signature verification of batch unaggregate processing"
+    );
+    pub static ref ATTESTATION_PROCESSING_BATCH_UNAGG_SIGNATURE_TIMES: Result<Histogram> = try_create_histogram(
+        "beacon_attestation_processing_batch_unagg_signature_times",
+        "Time spent on the signature verification of batch unaggregate attestation processing"
+    );
+
     /*
      * Shuffling cache
      */