Use blocking task for DB op

Author: michaelsproul

validator_client/lighthouse_validator_store/src/lib.rs

@@ -883,7 +883,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
     }
 
     async fn sign_attestations(
-        &self,
+        self: &Arc<Self>,
         mut attestations: Vec<(PublicKeyBytes, usize, Attestation<Self::E>)>,
     ) -> Result<Vec<Attestation<E>>, Error> {
         // Sign all attestations concurrently.
@@ -934,8 +934,18 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
             return Ok(vec![]);
         }
 
-        // Check slashing protection and insert into database.
-        let safe_attestations = self.slashing_protect_attestations(signed_attestations)?;
+        // Check slashing protection and insert into database. Use a dedicated blocking thread
+        // to avoid clogging the async executor with blocking database I/O.
+        let validator_store = self.clone();
+        let safe_attestations = self
+            .task_executor
+            .spawn_blocking_handle(
+                move || validator_store.slashing_protect_attestations(signed_attestations),
+                "slashing_protect_attestations",
+            )
+            .ok_or(Error::ExecutorError)?
+            .await
+            .map_err(|_| Error::ExecutorError)??;
         Ok(safe_attestations)
     }
 

validator_client/validator_store/src/lib.rs

@@ -22,6 +22,7 @@ pub enum Error<T> {
     GreaterThanCurrentEpoch { epoch: Epoch, current_epoch: Epoch },
     UnableToSignAttestation(AttestationError),
     SpecificError(T),
+    ExecutorError,
     Middleware(String),
 }
 
@@ -108,7 +109,7 @@ pub trait ValidatorStore: Send + Sync {
     /// Only successfully signed attestations that pass slashing protection are returned.
     #[allow(clippy::type_complexity)]
     fn sign_attestations(
-        &self,
+        self: &Arc<Self>,
         attestations: Vec<(PublicKeyBytes, usize, Attestation<Self::E>)>,
     ) -> impl Future<Output = Result<Vec<Attestation<Self::E>>, Error<Self::Error>>> + Send;