Check slashability of attestations in batches to avoid sequential bottleneck

validator_client/http_api/src/tests/keystores.rs

@@ -1107,6 +1107,12 @@ async fn generic_migration_test(
                 .sign_attestation(public_key, 0, &mut attestation, current_epoch)
                 .await
                 .unwrap();
+            let safe_attestations = tester1
+                .validator_store
+                .check_and_insert_attestations(vec![(attestation.clone(), public_key)])
+                .unwrap();
+            assert_eq!(safe_attestations.len(), 1);
+            assert_eq!(safe_attestations, vec![(attestation, public_key)]);
         }
 
         // Delete the selected keys from VC1.
@@ -1181,13 +1187,24 @@ async fn generic_migration_test(
         for (validator_index, mut attestation, should_succeed) in second_vc_attestations {
             let public_key = keystore_pubkey(&keystores[validator_index]);
             let current_epoch = attestation.data().target.epoch;
-            match tester2
+            if tester2
                 .validator_store
                 .sign_attestation(public_key, 0, &mut attestation, current_epoch)
                 .await
+                .is_err()
             {
-                Ok(()) => assert!(should_succeed),
-                Err(e) => assert!(!should_succeed, "{:?}", e),
+                // Doppelganger protected.
+                assert!(!should_succeed);
+                continue;
+            }
+            let safe_attestations = tester2
+                .validator_store
+                .check_and_insert_attestations(vec![(attestation.clone(), public_key)])
+                .unwrap();
+            if should_succeed {
+                assert_eq!(safe_attestations[0], (attestation, public_key));
+            } else {
+                assert!(safe_attestations.is_empty());
             }
         }
     })

validator_client/lighthouse_validator_store/src/lib.rs

@@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize};
 use signing_method::Error as SigningError;
 use signing_method::{SignableMessage, SigningContext, SigningMethod};
 use slashing_protection::{
-    InterchangeError, NotSafe, Safe, SlashingDatabase, interchange::Interchange,
+    CheckSlashability, InterchangeError, NotSafe, Safe, SlashingDatabase, interchange::Interchange,
 };
 use slot_clock::SlotClock;
 use std::marker::PhantomData;
@@ -766,77 +766,114 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         // Get the signing method and check doppelganger protection.
         let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;
 
-        // Checking for slashing conditions.
+        // Sign the attestation.
         let signing_epoch = attestation.data().target.epoch;
         let signing_context = self.signing_context(Domain::BeaconAttester, signing_epoch);
-        let domain_hash = signing_context.domain_hash(&self.spec);
-        let slashing_status = if signing_method
-            .requires_local_slashing_protection(self.enable_web3signer_slashing_protection)
-        {
-            self.slashing_protection.check_and_insert_attestation(
-                &validator_pubkey,
-                attestation.data(),
-                domain_hash,
+
+        let signature = signing_method
+            .get_signature::<E, BlindedPayload<E>>(
+                SignableMessage::AttestationData(attestation.data()),
+                signing_context,
+                &self.spec,
+                &self.task_executor,
             )
-        } else {
-            Ok(Safe::Valid)
-        };
+            .await?;
+        attestation
+            .add_signature(&signature, validator_committee_position)
+            .map_err(Error::UnableToSignAttestation)?;
 
-        match slashing_status {
-            // We can safely sign this attestation.
-            Ok(Safe::Valid) => {
-                let signature = signing_method
-                    .get_signature::<E, BlindedPayload<E>>(
-                        SignableMessage::AttestationData(attestation.data()),
-                        signing_context,
-                        &self.spec,
-                        &self.task_executor,
-                    )
-                    .await?;
-                attestation
-                    .add_signature(&signature, validator_committee_position)
-                    .map_err(Error::UnableToSignAttestation)?;
+        Ok(())
+    }
 
-                validator_metrics::inc_counter_vec(
-                    &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
-                    &[validator_metrics::SUCCESS],
-                );
+    #[instrument(
+        name = "store_check_and_insert_attestations",
+        level = "debug",
+        skip_all
+    )]
+    fn check_and_insert_attestations(
+        &self,
+        attestations: Vec<(Attestation<E>, PublicKeyBytes)>,
+    ) -> Result<Vec<(Attestation<E>, PublicKeyBytes)>, Error> {
+        let mut safe_attestations = vec![];
+        let mut attestations_to_check = vec![];
+
+        // Split attestations into de-facto safe attestations (checked by web3signer's slashing
+        // protection) and ones requiring checking against the slashing protection DB.
+        for (attestation, validator_pubkey) in &attestations {
+            let signing_method = self.doppelganger_checked_signing_method(*validator_pubkey)?;
+            let signing_epoch = attestation.data().target.epoch;
+            let signing_context = self.signing_context(Domain::BeaconAttester, signing_epoch);
+            let domain_hash = signing_context.domain_hash(&self.spec);
+
+            let check_slashability = if signing_method
+                .requires_local_slashing_protection(self.enable_web3signer_slashing_protection)
+            {
+                CheckSlashability::Yes
+            } else {
+                CheckSlashability::No
+            };
+            attestations_to_check.push((
+                attestation.data(),
+                validator_pubkey,
+                domain_hash,
+                check_slashability,
+            ));
+        }
 
-                Ok(())
-            }
-            Ok(Safe::SameData) => {
-                warn!("Skipping signing of previously signed attestation");
-                validator_metrics::inc_counter_vec(
-                    &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
-                    &[validator_metrics::SAME_DATA],
-                );
-                Err(Error::SameData)
-            }
-            Err(NotSafe::UnregisteredValidator(pk)) => {
-                warn!(
-                    msg = "Carefully consider running with --init-slashing-protection (see --help)",
-                    public_key = format!("{:?}", pk),
-                    "Not signing attestation for unregistered validator"
-                );
-                validator_metrics::inc_counter_vec(
-                    &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
-                    &[validator_metrics::UNREGISTERED],
-                );
-                Err(Error::Slashable(NotSafe::UnregisteredValidator(pk)))
-            }
-            Err(e) => {
-                crit!(
-                    attestation = format!("{:?}", attestation.data()),
-                    error = format!("{:?}", e),
-                    "Not signing slashable attestation"
-                );
-                validator_metrics::inc_counter_vec(
-                    &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
-                    &[validator_metrics::SLASHABLE],
-                );
-                Err(Error::Slashable(e))
+        // Batch check the attestations against the slashing protection DB while preserving the
+        // order so we can zip the results against the original vec.
+        //
+        // If the DB transaction fails then we consider the entire batch slashable and discard it.
+        let results = self
+            .slashing_protection
+            .check_and_insert_attestations(&attestations_to_check)
+            .map_err(Error::Slashable)?;
+
+        for ((attestation, validator_pubkey), slashing_status) in
+            attestations.into_iter().zip(results.into_iter())
+        {
+            match slashing_status {
+                Ok(Safe::Valid) => {
+                    safe_attestations.push((attestation, validator_pubkey));
+                    validator_metrics::inc_counter_vec(
+                        &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
+                        &[validator_metrics::SUCCESS],
+                    );
+                }
+                Ok(Safe::SameData) => {
+                    warn!("Skipping previously signed attestation");
+                    validator_metrics::inc_counter_vec(
+                        &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
+                        &[validator_metrics::SAME_DATA],
+                    );
+                }
+                Err(NotSafe::UnregisteredValidator(pk)) => {
+                    warn!(
+                        msg = "Carefully consider running with --init-slashing-protection (see --help)",
+                        public_key = ?pk,
+                        "Not signing attestation for unregistered validator"
+                    );
+                    validator_metrics::inc_counter_vec(
+                        &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
+                        &[validator_metrics::UNREGISTERED],
+                    );
+                }
+                Err(e) => {
+                    // FIXME(sproul): remove attestation data + make this error less scary
+                    crit!(
+                        attestation = format!("{:?}", attestation.data()),
+                        error = format!("{:?}", e),
+                        "Not signing slashable attestation"
+                    );
+                    validator_metrics::inc_counter_vec(
+                        &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
+                        &[validator_metrics::SLASHABLE],
+                    );
+                }
             }
         }
+
+        Ok(safe_attestations)
     }
 
     async fn sign_validator_registration_data(

validator_client/slashing_protection/src/interchange_test.rs

@@ -135,12 +135,15 @@ impl MultiTestCase {
             }
 
             for (i, att) in test_case.attestations.iter().enumerate() {
-                match slashing_db.check_and_insert_attestation_signing_root(
-                    &att.pubkey,
-                    att.source_epoch,
-                    att.target_epoch,
-                    SigningRoot::from(att.signing_root),
-                ) {
+                match slashing_db.with_transaction(|txn| {
+                    slashing_db.check_and_insert_attestation_signing_root(
+                        &att.pubkey,
+                        att.source_epoch,
+                        att.target_epoch,
+                        SigningRoot::from(att.signing_root),
+                        txn,
+                    )
+                }) {
                     Ok(safe) if !att.should_succeed => {
                         panic!(
                             "attestation {} from `{}` succeeded when it should have failed: {:?}",

validator_client/slashing_protection/src/lib.rs

@@ -16,8 +16,8 @@ pub mod interchange {
 pub use crate::signed_attestation::{InvalidAttestation, SignedAttestation};
 pub use crate::signed_block::{InvalidBlock, SignedBlock};
 pub use crate::slashing_database::{
-    InterchangeError, InterchangeImportOutcome, SUPPORTED_INTERCHANGE_FORMAT_VERSION,
-    SlashingDatabase,
+    CheckSlashability, InterchangeError, InterchangeImportOutcome,
+    SUPPORTED_INTERCHANGE_FORMAT_VERSION, SlashingDatabase,
 };
 use bls::PublicKeyBytes;
 use rusqlite::Error as SQLError;

validator_client/slashing_protection/src/parallel_tests.rs

@@ -44,11 +44,14 @@ fn attestation_same_target() {
     let results = (0..num_attestations)
         .into_par_iter()
         .map(|i| {
-            slashing_db.check_and_insert_attestation(
-                &pk,
-                &attestation_data_builder(i, num_attestations),
-                DEFAULT_DOMAIN,
-            )
+            slashing_db.with_transaction(|txn| {
+                slashing_db.check_and_insert_attestation(
+                    &pk,
+                    &attestation_data_builder(i, num_attestations),
+                    DEFAULT_DOMAIN,
+                    txn,
+                )
+            })
         })
         .collect::<Vec<_>>();
 
@@ -73,7 +76,9 @@ fn attestation_surround_fest() {
         .into_par_iter()
         .map(|i| {
             let att = attestation_data_builder(i, 2 * num_attestations - i);
-            slashing_db.check_and_insert_attestation(&pk, &att, DEFAULT_DOMAIN)
+            slashing_db.with_transaction(|txn| {
+                slashing_db.check_and_insert_attestation(&pk, &att, DEFAULT_DOMAIN, txn)
+            })
         })
         .collect::<Vec<_>>();
 

validator_client/slashing_protection/src/slashing_database.rs

@@ -38,6 +38,17 @@ pub struct SlashingDatabase {
     conn_pool: Pool,
 }
 
+/// Whether to check slashability of a message.
+///
+/// The `No` variant MUST only be used if there is another source of slashing protection configured,
+/// e.g. web3signer's slashing protection.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum CheckSlashability {
+    #[default]
+    Yes,
+    No,
+}
+
 impl SlashingDatabase {
     /// Open an existing database at the given `path`, or create one if none exists.
     pub fn open_or_create(path: &Path) -> Result<Self, NotSafe> {
@@ -635,6 +646,43 @@ impl SlashingDatabase {
         self.check_block_proposal(&txn, validator_pubkey, slot, signing_root)
     }
 
+    #[instrument(name = "db_check_and_insert_attestations", level = "debug", skip_all)]
+    pub fn check_and_insert_attestations<'a>(
+        &self,
+        attestations: &'a [(
+            &'a AttestationData,
+            &'a PublicKeyBytes,
+            Hash256,
+            CheckSlashability,
+        )],
+    ) -> Result<Vec<Result<Safe, NotSafe>>, NotSafe> {
+        let mut conn = self.conn_pool.get()?;
+        let txn = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
+
+        let mut results = vec![];
+        for (attestation, validator_pubkey, domain, check_slashability) in attestations {
+            match check_slashability {
+                CheckSlashability::No => {
+                    results.push(Ok(Safe::Valid));
+                }
+                CheckSlashability::Yes => {
+                    let attestation_signing_root = attestation.signing_root(*domain).into();
+                    results.push(self.check_and_insert_attestation_signing_root(
+                        validator_pubkey,
+                        attestation.source.epoch,
+                        attestation.target.epoch,
+                        attestation_signing_root,
+                        &txn,
+                    ));
+                }
+            }
+        }
+
+        txn.commit()?;
+
+        Ok(results)
+    }
+
     /// Check an attestation for slash safety, and if it is safe, record it in the database.
     ///
     /// The checking and inserting happen atomically and exclusively. We enforce exclusivity
@@ -647,13 +695,15 @@ impl SlashingDatabase {
         validator_pubkey: &PublicKeyBytes,
         attestation: &AttestationData,
         domain: Hash256,
+        txn: &Transaction,
     ) -> Result<Safe, NotSafe> {
         let attestation_signing_root = attestation.signing_root(domain).into();
         self.check_and_insert_attestation_signing_root(
             validator_pubkey,
             attestation.source.epoch,
             attestation.target.epoch,
             attestation_signing_root,
+            txn,
         )
     }
 
@@ -664,17 +714,15 @@ impl SlashingDatabase {
         att_source_epoch: Epoch,
         att_target_epoch: Epoch,
         att_signing_root: SigningRoot,
+        txn: &Transaction,
     ) -> Result<Safe, NotSafe> {
-        let mut conn = self.conn_pool.get()?;
-        let txn = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
         let safe = self.check_and_insert_attestation_signing_root_txn(
             validator_pubkey,
             att_source_epoch,
             att_target_epoch,
             att_signing_root,
-            &txn,
+            txn,
         )?;
-        txn.commit()?;
         Ok(safe)
     }