Attestation batch new refactor

testing/web3signer_tests/src/lib.rs

@@ -605,12 +605,13 @@ mod tests {
         })
         .await
         .assert_signatures_match("attestation", |pubkey, validator_store| async move {
-            let mut attestation = get_attestation();
+            let attestation = get_attestation();
             validator_store
-                .sign_attestation(pubkey, 0, &mut attestation, Epoch::new(0))
+                .sign_attestations(vec![(pubkey, 0, attestation)])
                 .await
-                .unwrap();
-            attestation
+                .unwrap()
+                .pop()
+                .unwrap()
         })
         .await
         .assert_signatures_match("signed_aggregate", |pubkey, validator_store| async move {
@@ -820,8 +821,6 @@ mod tests {
             block
         };
 
-        let current_epoch = Epoch::new(5);
-
         TestingRig::new(
             network,
             slashing_protection_config,
@@ -830,43 +829,47 @@ mod tests {
         )
         .await
         .assert_signatures_match("first_attestation", |pubkey, validator_store| async move {
-            let mut attestation = first_attestation();
+            let attestation = first_attestation();
             validator_store
-                .sign_attestation(pubkey, 0, &mut attestation, current_epoch)
+                .sign_attestations(vec![(pubkey, 0, attestation)])
                 .await
-                .unwrap();
-            attestation
+                .unwrap()
+                .pop()
+                .unwrap()
         })
         .await
         .assert_slashable_message_should_sign(
             "double_vote_attestation",
             move |pubkey, validator_store| async move {
-                let mut attestation = double_vote_attestation();
+                let attestation = double_vote_attestation();
                 validator_store
-                    .sign_attestation(pubkey, 0, &mut attestation, current_epoch)
+                    .sign_attestations(vec![(pubkey, 0, attestation)])
                     .await
+                    .map(|_| ())
             },
             slashable_message_should_sign,
         )
         .await
         .assert_slashable_message_should_sign(
             "surrounding_attestation",
             move |pubkey, validator_store| async move {
-                let mut attestation = surrounding_attestation();
+                let attestation = surrounding_attestation();
                 validator_store
-                    .sign_attestation(pubkey, 0, &mut attestation, current_epoch)
+                    .sign_attestations(vec![(pubkey, 0, attestation)])
                     .await
+                    .map(|_| ())
             },
             slashable_message_should_sign,
         )
         .await
         .assert_slashable_message_should_sign(
             "surrounded_attestation",
             move |pubkey, validator_store| async move {
-                let mut attestation = surrounded_attestation();
+                let attestation = surrounded_attestation();
                 validator_store
-                    .sign_attestation(pubkey, 0, &mut attestation, current_epoch)
+                    .sign_attestations(vec![(pubkey, 0, attestation)])
                     .await
+                    .map(|_| ())
             },
             slashable_message_should_sign,
         )

validator_client/http_api/src/tests/keystores.rs

@@ -1099,20 +1099,15 @@ async fn generic_migration_test(
         check_keystore_import_response(&import_res, all_imported(keystores.len()));
 
         // Sign attestations on VC1.
-        for (validator_index, mut attestation) in first_vc_attestations {
+        for (validator_index, attestation) in first_vc_attestations {
             let public_key = keystore_pubkey(&keystores[validator_index]);
-            let current_epoch = attestation.data().target.epoch;
-            tester1
-                .validator_store
-                .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)])
+                .sign_attestations(vec![(public_key, 0, attestation.clone())])
+                .await
                 .unwrap();
             assert_eq!(safe_attestations.len(), 1);
-            assert_eq!(safe_attestations, vec![(attestation, public_key)]);
+            assert_eq!(safe_attestations, vec![attestation]);
         }
 
         // Delete the selected keys from VC1.
@@ -1184,27 +1179,24 @@ async fn generic_migration_test(
         check_keystore_import_response(&import_res, all_imported(import_indices.len()));
 
         // Sign attestations on the second VC.
-        for (validator_index, mut attestation, should_succeed) in second_vc_attestations {
+        for (validator_index, attestation, should_succeed) in second_vc_attestations {
             let public_key = keystore_pubkey(&keystores[validator_index]);
-            let current_epoch = attestation.data().target.epoch;
-            if tester2
-                .validator_store
-                .sign_attestation(public_key, 0, &mut attestation, current_epoch)
-                .await
-                .is_err()
-            {
-                // Doppelganger protected.
-                assert!(!should_succeed);
-                continue;
-            }
-            let safe_attestations = tester2
+            let result = 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());
+                .sign_attestations(vec![(public_key, 0, attestation.clone())])
+                .await;
+            match result {
+                Ok(safe_attestations) => {
+                    if should_succeed {
+                        assert_eq!(safe_attestations, vec![attestation]);
+                    } else {
+                        assert!(safe_attestations.is_empty());
+                    }
+                }
+                Err(_) => {
+                    // Doppelganger protected or other error.
+                    assert!(!should_succeed);
+                }
             }
         }
     })
@@ -1330,10 +1322,10 @@ async fn delete_concurrent_with_signing() {
 
         let handle = handle.spawn(async move {
             for j in 0..num_attestations {
-                let mut att = make_attestation(j, j + 1);
+                let att = make_attestation(j, j + 1);
                 for public_key in thread_pubkeys.iter() {
                     let _ = validator_store
-                        .sign_attestation(*public_key, 0, &mut att, Epoch::new(j + 1))
+                        .sign_attestations(vec![(*public_key, 0, att.clone())])
                         .await;
                 }
             }

validator_client/lighthouse_validator_store/Cargo.toml

@@ -12,6 +12,7 @@ doppelganger_service = { workspace = true }
 either = { workspace = true }
 environment = { workspace = true }
 eth2 = { workspace = true }
+futures = { workspace = true }
 initialized_validators = { workspace = true }
 logging = { workspace = true }
 parking_lot = { workspace = true }

validator_client/lighthouse_validator_store/src/lib.rs

@@ -2,6 +2,7 @@ use account_utils::validator_definitions::{PasswordStorage, ValidatorDefinition}
 use bls::{PublicKeyBytes, Signature};
 use doppelganger_service::DoppelgangerService;
 use eth2::types::PublishBlockRequest;
+use futures::future::join_all;
 use initialized_validators::InitializedValidators;
 use logging::crit;
 use parking_lot::{Mutex, RwLock};
@@ -747,22 +748,71 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         }
     }
 
+    async fn sign_attestations(
+        &self,
+        mut attestations: Vec<(PublicKeyBytes, usize, Attestation<Self::E>)>,
+    ) -> Result<Vec<Attestation<E>>, Error> {
+        // Sign all attestations concurrently.
+        let signing_futures =
+            attestations
+                .iter_mut()
+                .map(|(pubkey, validator_committee_index, attestation)| {
+                    let pubkey = *pubkey;
+                    let validator_committee_index = *validator_committee_index;
+                    async move {
+                        self.sign_attestation_no_checks(
+                            pubkey,
+                            validator_committee_index,
+                            attestation,
+                        )
+                        .await
+                        .map(|_| pubkey)
+                    }
+                });
+
+        // Execute all signing in parallel.
+        let results: Vec<_> = join_all(signing_futures).await;
+
+        // Collect successfully signed attestations and log errors.
+        let mut signed_attestations = Vec::new();
+        for (result, (pubkey, _, attestation)) in results.into_iter().zip(attestations.into_iter())
+        {
+            match result {
+                Ok(_) => {
+                    signed_attestations.push((attestation, pubkey));
+                }
+                Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
+                    warn!(
+                        info = "a validator may have recently been removed from this VC",
+                        ?pubkey,
+                        "Missing pubkey for attestation"
+                    );
+                }
+                Err(e) => {
+                    crit!(
+                        error = ?e,
+                        "Failed to sign attestation"
+                    );
+                }
+            }
+        }
+
+        if signed_attestations.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        // Check slashing protection and insert into database.
+        let safe_attestations = self.check_and_insert_attestations(signed_attestations)?;
+        Ok(safe_attestations.into_iter().map(|(a, _)| a).collect())
+    }
+
     #[instrument(skip_all)]
-    async fn sign_attestation(
+    async fn sign_attestation_no_checks(
         &self,
         validator_pubkey: PublicKeyBytes,
         validator_committee_position: usize,
         attestation: &mut Attestation<E>,
-        current_epoch: Epoch,
     ) -> Result<(), Error> {
-        // Make sure the target epoch is not higher than the current epoch to avoid potential attacks.
-        if attestation.data().target.epoch > current_epoch {
-            return Err(Error::GreaterThanCurrentEpoch {
-                epoch: attestation.data().target.epoch,
-                current_epoch,
-            });
-        }
-
         // Get the signing method and check doppelganger protection.
         let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;