Self review changes

Author: michaelsproul

common/task_executor/src/rayon_pool_provider.rs

@@ -15,7 +15,7 @@ pub struct RayonPoolProvider {
     /// By default ~25% of CPUs or a minimum of 1 thread.
     low_priority_thread_pool: Arc<ThreadPool>,
     /// Larger rayon thread pool for high-priority, compute-intensive tasks.
-    /// By default ~80% of CPUs or a minimum of 1 thread. Citical/highest
+    /// By default ~80% of CPUs or a minimum of 1 thread. Critical/highest
     /// priority tasks should use the global pool instead.
     high_priority_thread_pool: Arc<ThreadPool>,
 }

validator_client/http_api/src/tests/keystores.rs

@@ -1109,6 +1109,8 @@ async fn generic_migration_test(
             assert_eq!(safe_attestations.len(), 1);
             // Compare data only, ignoring signatures which are added during signing.
             assert_eq!(safe_attestations[0].data(), attestation.data());
+            // Check that the signature is non-zero.
+            assert!(!safe_attestations[0].signature().is_infinity());
         }
 
         // Delete the selected keys from VC1.
@@ -1192,6 +1194,8 @@ async fn generic_migration_test(
                         // Compare data only, ignoring signatures which are added during signing.
                         assert_eq!(safe_attestations.len(), 1);
                         assert_eq!(safe_attestations[0].data(), attestation.data());
+                        // Check that the signature is non-zero.
+                        assert!(!safe_attestations[0].signature().is_infinity());
                     } else {
                         assert!(safe_attestations.is_empty());
                     }

validator_client/lighthouse_validator_store/src/lib.rs

@@ -558,8 +558,14 @@ impl<T: SlotClock + 'static, E: EthSpec> LighthouseValidatorStore<T, E> {
         })
     }
 
-    #[instrument(skip_all)]
-    async fn sign_attestation_no_checks(
+    /// Sign an attestation without performing any slashing protection checks.
+    ///
+    /// THIS METHOD IS DANGEROUS AND SHOULD ONLY BE USED INTERNALLY IMMEDIATELY PRIOR TO A
+    /// SLASHING PROTECTION CHECK. See `slashing_protect_attestations`.
+    ///
+    /// This method DOES perform doppelganger protection checks.
+    #[instrument(level = "debug", skip_all)]
+    async fn sign_attestation_no_slashing_protection(
         &self,
         validator_pubkey: PublicKeyBytes,
         validator_committee_position: usize,
@@ -587,20 +593,27 @@ impl<T: SlotClock + 'static, E: EthSpec> LighthouseValidatorStore<T, E> {
         Ok(())
     }
 
-    #[instrument(
-        name = "store_check_and_insert_attestations",
-        level = "debug",
-        skip_all
-    )]
-    fn check_and_insert_attestations(
+    /// Provide slashing protection for `attestations`, safely updating the slashing protection DB.
+    ///
+    /// Return a vec of safe attestations which have passed slashing protection. Unsafe attestations
+    /// will be dropped and result in warning logs.
+    ///
+    /// This method SKIPS slashing protection for web3signer validators that have slashing
+    /// protection disabled at the Lighthouse layer. It is up to the user to ensure slashing
+    /// protection is enabled in web3signer instead.
+    #[instrument(level = "debug", skip_all)]
+    fn slashing_protect_attestations(
         &self,
         attestations: Vec<(Attestation<E>, PublicKeyBytes)>,
-    ) -> Result<Vec<(Attestation<E>, PublicKeyBytes)>, Error> {
-        let mut safe_attestations = vec![];
-        let mut attestations_to_check = vec![];
+    ) -> Result<Vec<Attestation<E>>, Error> {
+        let mut safe_attestations = Vec::with_capacity(attestations.len());
+        let mut attestations_to_check = Vec::with_capacity(attestations.len());
 
         // Split attestations into de-facto safe attestations (checked by web3signer's slashing
         // protection) and ones requiring checking against the slashing protection DB.
+        //
+        // All attestations are added to `attestation_to_check`, with skipped attestations having
+        // `CheckSlashability::No`.
         for (attestation, validator_pubkey) in &attestations {
             let signing_method = self.doppelganger_checked_signing_method(*validator_pubkey)?;
             let signing_epoch = attestation.data().target.epoch;
@@ -636,7 +649,7 @@ impl<T: SlotClock + 'static, E: EthSpec> LighthouseValidatorStore<T, E> {
         {
             match slashing_status {
                 Ok(Safe::Valid) => {
-                    safe_attestations.push((attestation, validator_pubkey));
+                    safe_attestations.push(attestation);
                     validator_metrics::inc_counter_vec(
                         &validator_metrics::SIGNED_ATTESTATIONS_TOTAL,
                         &[validator_metrics::SUCCESS],
@@ -881,25 +894,24 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
                     let pubkey = *pubkey;
                     let validator_committee_index = *validator_committee_index;
                     async move {
-                        self.sign_attestation_no_checks(
+                        self.sign_attestation_no_slashing_protection(
                             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();
+        let mut signed_attestations = Vec::with_capacity(attestations.len());
         for (result, (pubkey, _, attestation)) in results.into_iter().zip(attestations.into_iter())
         {
             match result {
-                Ok(_) => {
+                Ok(()) => {
                     signed_attestations.push((attestation, pubkey));
                 }
                 Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
@@ -919,12 +931,12 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         }
 
         if signed_attestations.is_empty() {
-            return Ok(Vec::new());
+            return Ok(vec![]);
         }
 
         // 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())
+        let safe_attestations = self.slashing_protect_attestations(signed_attestations)?;
+        Ok(safe_attestations)
     }
 
     async fn sign_validator_registration_data(

validator_client/signing_method/src/lib.rs

@@ -181,6 +181,10 @@ impl SigningMethod {
                 let voting_keypair = voting_keypair.clone();
                 // Spawn a blocking task to produce the signature. This avoids blocking the core
                 // tokio executor.
+                //
+                // We are using the Rayon high-priority pool which uses up to 80% of available
+                // threads. In future we could consider using 90-100% in the VC, seeing as we have
+                // very little other work to do aside from signing.
                 let signature = executor
                     .spawn_blocking_with_rayon_async(RayonPoolType::HighPriority, move || {
                         voting_keypair.sk.sign(signing_root)

validator_client/slashing_protection/src/slashing_database.rs

@@ -194,7 +194,9 @@ impl SlashingDatabase {
         U: From<NotSafe>,
     {
         let mut conn = self.conn_pool.get().map_err(NotSafe::from)?;
-        let txn = conn.transaction().map_err(NotSafe::from)?;
+        let txn = conn
+            .transaction_with_behavior(TransactionBehavior::Exclusive)
+            .map_err(NotSafe::from)?;
         let value = f(&txn)?;
         txn.commit().map_err(NotSafe::from)?;
         Ok(value)
@@ -659,7 +661,7 @@ impl SlashingDatabase {
         let mut conn = self.conn_pool.get()?;
         let txn = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
 
-        let mut results = vec![];
+        let mut results = Vec::with_capacity(attestations.len());
         for (attestation, validator_pubkey, domain, check_slashability) in attestations {
             match check_slashability {
                 CheckSlashability::No => {

validator_client/validator_store/src/lib.rs

@@ -103,6 +103,9 @@ pub trait ValidatorStore: Send + Sync {
         current_slot: Slot,
     ) -> impl Future<Output = Result<SignedBlock<Self::E>, Error<Self::Error>>> + Send;
 
+    /// Sign a batch of `attestations` and apply slashing protection to them.
+    ///
+    /// Only successfully signed attestations that pass slashing protection are returned.
     #[allow(clippy::type_complexity)]
     fn sign_attestations(
         &self,