Refactor/stream vc vote publishing (#8880)

Changes four `ValidatorStore` batch signing methods to return `impl Stream` instead of `Future`. Services consume the stream and publish each batch as it arrives.  No behavioral change for lh since `LighthouseValidatorStore` wraps everything in `stream::once`

Also replaces anonymous tuples in method signatures with named structs


Co-Authored-By: shane-moore <skm1790@gmail.com>

Co-Authored-By: Michael Sproul <michaelsproul@users.noreply.github.com>

Co-Authored-By: Mac L <mjladson@pm.me>

Cargo.lock

@@ -9710,6 +9710,7 @@ version = "0.1.0"
 dependencies = [
  "bls",
  "eth2",
+ "futures",
  "slashing_protection",
  "types",
 ]

testing/web3signer_tests/src/lib.rs

@@ -25,6 +25,7 @@ mod tests {
     use eth2_keystore::KeystoreBuilder;
     use eth2_network_config::Eth2NetworkConfig;
     use fixed_bytes::FixedBytesExtended;
+    use futures::StreamExt;
     use initialized_validators::{
         InitializedValidators, load_pem_certificate, load_pkcs12_identity,
     };
@@ -50,7 +51,7 @@ mod tests {
     use types::{attestation::AttestationBase, *};
     use url::Url;
     use validator_store::{
-        Error as ValidatorStoreError, SignedBlock, UnsignedBlock, ValidatorStore,
+        AttestationToSign, Error as ValidatorStoreError, SignedBlock, UnsignedBlock, ValidatorStore,
     };
 
     /// If the we are unable to reach the Web3Signer HTTP API within this time out then we will
@@ -654,13 +655,14 @@ mod tests {
         .await
         .assert_signatures_match("attestation", |pubkey, validator_store| async move {
             let attestation = get_attestation();
-            validator_store
+            let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                .sign_attestations(vec![(0, pubkey, 0, attestation)])
+                validator_index: 0,
-                .await
+                pubkey,
-                .unwrap()
+                validator_committee_index: 0,
-                .pop()
+                attestation,
-                .unwrap()
+            }]);
-                .1
+            tokio::pin!(stream);
+            stream.next().await.unwrap().unwrap().pop().unwrap().1
         })
         .await
         .assert_signatures_match("signed_aggregate", |pubkey, validator_store| async move {
@@ -879,22 +881,28 @@ mod tests {
         .await
         .assert_signatures_match("first_attestation", |pubkey, validator_store| async move {
             let attestation = first_attestation();
-            validator_store
+            let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                .sign_attestations(vec![(0, pubkey, 0, attestation)])
+                validator_index: 0,
-                .await
+                pubkey,
-                .unwrap()
+                validator_committee_index: 0,
-                .pop()
+                attestation,
-                .unwrap()
+            }]);
-                .1
+            tokio::pin!(stream);
+            stream.next().await.unwrap().unwrap().pop().unwrap().1
         })
         .await
         .assert_slashable_attestation_should_sign(
             "double_vote_attestation",
             move |pubkey, validator_store| async move {
                 let attestation = double_vote_attestation();
-                validator_store
+                let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                    .sign_attestations(vec![(0, pubkey, 0, attestation)])
+                    validator_index: 0,
-                    .await
+                    pubkey,
+                    validator_committee_index: 0,
+                    attestation,
+                }]);
+                tokio::pin!(stream);
+                stream.next().await.unwrap()
             },
             slashable_message_should_sign,
         )
@@ -903,9 +911,14 @@ mod tests {
             "surrounding_attestation",
             move |pubkey, validator_store| async move {
                 let attestation = surrounding_attestation();
-                validator_store
+                let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                    .sign_attestations(vec![(0, pubkey, 0, attestation)])
+                    validator_index: 0,
-                    .await
+                    pubkey,
+                    validator_committee_index: 0,
+                    attestation,
+                }]);
+                tokio::pin!(stream);
+                stream.next().await.unwrap()
             },
             slashable_message_should_sign,
         )
@@ -914,9 +927,14 @@ mod tests {
             "surrounded_attestation",
             move |pubkey, validator_store| async move {
                 let attestation = surrounded_attestation();
-                validator_store
+                let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                    .sign_attestations(vec![(0, pubkey, 0, attestation)])
+                    validator_index: 0,
-                    .await
+                    pubkey,
+                    validator_committee_index: 0,
+                    attestation,
+                }]);
+                tokio::pin!(stream);
+                stream.next().await.unwrap()
             },
             slashable_message_should_sign,
         )

validator_client/http_api/src/tests/keystores.rs

@@ -9,6 +9,7 @@ use eth2::lighthouse_vc::{
     types::Web3SignerValidatorRequest,
 };
 use fixed_bytes::FixedBytesExtended;
+use futures::StreamExt;
 use itertools::Itertools;
 use lighthouse_validator_store::DEFAULT_GAS_LIMIT;
 use rand::rngs::StdRng;
@@ -19,6 +20,7 @@ use std::{collections::HashMap, path::Path};
 use tokio::runtime::Handle;
 use typenum::Unsigned;
 use types::{Address, attestation::AttestationBase};
+use validator_store::AttestationToSign;
 use validator_store::ValidatorStore;
 use zeroize::Zeroizing;
 
@@ -1101,11 +1103,16 @@ async fn generic_migration_test(
         // Sign attestations on VC1.
         for (validator_index, attestation) in first_vc_attestations {
             let public_key = keystore_pubkey(&keystores[validator_index]);
-            let safe_attestations = tester1
+            let stream = tester1
                 .validator_store
-                .sign_attestations(vec![(0, public_key, 0, attestation.clone())])
+                .sign_attestations(vec![AttestationToSign {
-                .await
+                    validator_index: 0,
-                .unwrap();
+                    pubkey: public_key,
+                    validator_committee_index: 0,
+                    attestation: attestation.clone(),
+                }]);
+            tokio::pin!(stream);
+            let safe_attestations = stream.next().await.unwrap().unwrap();
             assert_eq!(safe_attestations.len(), 1);
             // Compare data only, ignoring signatures which are added during signing.
             assert_eq!(safe_attestations[0].1.data(), attestation.data());
@@ -1184,10 +1191,16 @@ async fn generic_migration_test(
         // Sign attestations on the second VC.
         for (validator_index, attestation, should_succeed) in second_vc_attestations {
             let public_key = keystore_pubkey(&keystores[validator_index]);
-            let result = tester2
+            let stream = tester2
                 .validator_store
-                .sign_attestations(vec![(0, public_key, 0, attestation.clone())])
+                .sign_attestations(vec![AttestationToSign {
-                .await;
+                    validator_index: 0,
+                    pubkey: public_key,
+                    validator_committee_index: 0,
+                    attestation: attestation.clone(),
+                }]);
+            tokio::pin!(stream);
+            let result = stream.next().await.unwrap();
             match result {
                 Ok(safe_attestations) => {
                     if should_succeed {
@@ -1331,14 +1344,14 @@ async fn delete_concurrent_with_signing() {
             for j in 0..num_attestations {
                 let att = make_attestation(j, j + 1);
                 for (validator_index, public_key) in thread_pubkeys.iter().enumerate() {
-                    let _ = validator_store
+                    let stream = validator_store.sign_attestations(vec![AttestationToSign {
-                        .sign_attestations(vec![(
+                        validator_index: validator_index as u64,
-                            validator_index as u64,
+                        pubkey: *public_key,
-                            *public_key,
+                        validator_committee_index: 0,
-                            0,
+                        attestation: att.clone(),
-                            att.clone(),
+                    }]);
-                        )])
+                    tokio::pin!(stream);
-                        .await;
+                    let _ = stream.next().await;
                 }
             }
         });

validator_client/lighthouse_validator_store/src/lib.rs

@@ -2,7 +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 futures::{Stream, future::join_all, stream};
 use initialized_validators::InitializedValidators;
 use logging::crit;
 use parking_lot::{Mutex, RwLock};
@@ -17,7 +17,7 @@ use std::marker::PhantomData;
 use std::path::Path;
 use std::sync::Arc;
 use task_executor::TaskExecutor;
-use tracing::{error, info, instrument, warn};
+use tracing::{Instrument, debug, error, info, info_span, instrument, warn};
 use types::{
     AbstractExecPayload, Address, AggregateAndProof, Attestation, BeaconBlock, BlindedPayload,
     ChainSpec, ContributionAndProof, Domain, Epoch, EthSpec, ExecutionPayloadEnvelope, Fork,
@@ -28,7 +28,8 @@ use types::{
     ValidatorRegistrationData, VoluntaryExit, graffiti::GraffitiString,
 };
 use validator_store::{
-    DoppelgangerStatus, Error as ValidatorStoreError, ProposalData, SignedBlock, UnsignedBlock,
+    AggregateToSign, AttestationToSign, ContributionToSign, DoppelgangerStatus,
+    Error as ValidatorStoreError, ProposalData, SignedBlock, SyncMessageToSign, UnsignedBlock,
     ValidatorStore,
 };
 
@@ -691,6 +692,119 @@ impl<T: SlotClock + 'static, E: EthSpec> LighthouseValidatorStore<T, E> {
 
         Ok(safe_attestations)
     }
+
+    /// Signs an `AggregateAndProof` for a given validator.
+    ///
+    /// The resulting `SignedAggregateAndProof` is sent on the aggregation channel and cannot be
+    /// modified by actors other than the signing validator.
+    pub async fn produce_signed_aggregate_and_proof(
+        &self,
+        validator_pubkey: PublicKeyBytes,
+        aggregator_index: u64,
+        aggregate: Attestation<E>,
+        selection_proof: SelectionProof,
+    ) -> Result<SignedAggregateAndProof<E>, Error> {
+        let signing_epoch = aggregate.data().target.epoch;
+        let signing_context = self.signing_context(Domain::AggregateAndProof, signing_epoch);
+
+        let message =
+            AggregateAndProof::from_attestation(aggregator_index, aggregate, selection_proof);
+
+        let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;
+        let signature = signing_method
+            .get_signature::<E, BlindedPayload<E>>(
+                SignableMessage::SignedAggregateAndProof(message.to_ref()),
+                signing_context,
+                &self.spec,
+                &self.task_executor,
+            )
+            .await?;
+
+        validator_metrics::inc_counter_vec(
+            &validator_metrics::SIGNED_AGGREGATES_TOTAL,
+            &[validator_metrics::SUCCESS],
+        );
+
+        Ok(SignedAggregateAndProof::from_aggregate_and_proof(
+            message, signature,
+        ))
+    }
+
+    pub async fn produce_sync_committee_signature(
+        &self,
+        slot: Slot,
+        beacon_block_root: Hash256,
+        validator_index: u64,
+        validator_pubkey: &PublicKeyBytes,
+    ) -> Result<SyncCommitteeMessage, Error> {
+        let signing_epoch = slot.epoch(E::slots_per_epoch());
+        let signing_context = self.signing_context(Domain::SyncCommittee, signing_epoch);
+
+        // Bypass `with_validator_signing_method`: sync committee messages are not slashable.
+        let signing_method = self.doppelganger_bypassed_signing_method(*validator_pubkey)?;
+
+        let signature = signing_method
+            .get_signature::<E, BlindedPayload<E>>(
+                SignableMessage::SyncCommitteeSignature {
+                    beacon_block_root,
+                    slot,
+                },
+                signing_context,
+                &self.spec,
+                &self.task_executor,
+            )
+            .await
+            .map_err(Error::SpecificError)?;
+
+        validator_metrics::inc_counter_vec(
+            &validator_metrics::SIGNED_SYNC_COMMITTEE_MESSAGES_TOTAL,
+            &[validator_metrics::SUCCESS],
+        );
+
+        Ok(SyncCommitteeMessage {
+            slot,
+            beacon_block_root,
+            validator_index,
+            signature,
+        })
+    }
+
+    pub async fn produce_signed_contribution_and_proof(
+        &self,
+        aggregator_index: u64,
+        aggregator_pubkey: PublicKeyBytes,
+        contribution: SyncCommitteeContribution<E>,
+        selection_proof: SyncSelectionProof,
+    ) -> Result<SignedContributionAndProof<E>, Error> {
+        let signing_epoch = contribution.slot.epoch(E::slots_per_epoch());
+        let signing_context = self.signing_context(Domain::ContributionAndProof, signing_epoch);
+
+        // Bypass `with_validator_signing_method`: sync committee messages are not slashable.
+        let signing_method = self.doppelganger_bypassed_signing_method(aggregator_pubkey)?;
+
+        let message = ContributionAndProof {
+            aggregator_index,
+            contribution,
+            selection_proof: selection_proof.into(),
+        };
+
+        let signature = signing_method
+            .get_signature::<E, BlindedPayload<E>>(
+                SignableMessage::SignedContributionAndProof(&message),
+                signing_context,
+                &self.spec,
+                &self.task_executor,
+            )
+            .await
+            .map_err(Error::SpecificError)?;
+
+        validator_metrics::inc_counter_vec(
+            &validator_metrics::SIGNED_SYNC_COMMITTEE_CONTRIBUTIONS_TOTAL,
+            &[validator_metrics::SUCCESS],
+        );
+
+        Ok(SignedContributionAndProof { message, signature })
+    }
 }
 
 impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorStore<T, E> {
@@ -882,38 +996,48 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         }
     }
 
-    async fn sign_attestations(
+    fn sign_attestations(
         self: &Arc<Self>,
-        mut attestations: Vec<(u64, PublicKeyBytes, usize, Attestation<Self::E>)>,
+        mut attestations: Vec<AttestationToSign<E>>,
-    ) -> Result<Vec<(u64, Attestation<E>)>, Error> {
+    ) -> impl Stream<Item = Result<Vec<(u64, Attestation<E>)>, Error>> + Send {
+        let store = self.clone();
+        stream::once(async move {
             // Sign all attestations concurrently.
-        let signing_futures =
+            let signing_futures = attestations.iter_mut().map(
-            attestations
+                |AttestationToSign {
-                .iter_mut()
+                     pubkey,
-                .map(|(_, pubkey, validator_committee_index, attestation)| {
+                     validator_committee_index,
+                     attestation,
+                     ..
+                 }| {
                     let pubkey = *pubkey;
                     let validator_committee_index = *validator_committee_index;
+                    let store = store.clone();
                     async move {
-                        self.sign_attestation_no_slashing_protection(
+                        store
+                            .sign_attestation_no_slashing_protection(
                                 pubkey,
                                 validator_committee_index,
                                 attestation,
                             )
                             .await
                     }
-                });
+                },
+            );
 
             // 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::with_capacity(attestations.len());
-        for (result, (validator_index, pubkey, _, attestation)) in
+            for (result, att) in results.into_iter().zip(attestations.into_iter()) {
-            results.into_iter().zip(attestations.into_iter())
-        {
                 match result {
                     Ok(()) => {
-                    signed_attestations.push((validator_index, attestation, pubkey));
+                        signed_attestations.push((
+                            att.validator_index,
+                            att.attestation,
+                            att.pubkey,
+                        ));
                     }
                     Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
                         warn!(
@@ -935,10 +1059,10 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
                 return Ok(vec![]);
             }
 
-        // Check slashing protection and insert into database. Use a dedicated blocking thread
+            // Check slashing protection and insert into database. Use a dedicated blocking
-        // to avoid clogging the async executor with blocking database I/O.
+            // thread to avoid clogging the async executor with blocking database I/O.
-        let validator_store = self.clone();
+            let validator_store = store.clone();
-        let safe_attestations = self
+            let safe_attestations = store
                 .task_executor
                 .spawn_blocking_handle(
                     move || validator_store.slashing_protect_attestations(signed_attestations),
@@ -948,6 +1072,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
                 .await
                 .map_err(|_| Error::ExecutorError)??;
             Ok(safe_attestations)
+        })
     }
 
     async fn sign_validator_registration_data(
@@ -979,43 +1104,6 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         })
     }
 
-    /// Signs an `AggregateAndProof` for a given validator.
-    ///
-    /// The resulting `SignedAggregateAndProof` is sent on the aggregation channel and cannot be
-    /// modified by actors other than the signing validator.
-    async fn produce_signed_aggregate_and_proof(
-        &self,
-        validator_pubkey: PublicKeyBytes,
-        aggregator_index: u64,
-        aggregate: Attestation<E>,
-        selection_proof: SelectionProof,
-    ) -> Result<SignedAggregateAndProof<E>, Error> {
-        let signing_epoch = aggregate.data().target.epoch;
-        let signing_context = self.signing_context(Domain::AggregateAndProof, signing_epoch);
-
-        let message =
-            AggregateAndProof::from_attestation(aggregator_index, aggregate, selection_proof);
-
-        let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;
-        let signature = signing_method
-            .get_signature::<E, BlindedPayload<E>>(
-                SignableMessage::SignedAggregateAndProof(message.to_ref()),
-                signing_context,
-                &self.spec,
-                &self.task_executor,
-            )
-            .await?;
-
-        validator_metrics::inc_counter_vec(
-            &validator_metrics::SIGNED_AGGREGATES_TOTAL,
-            &[validator_metrics::SUCCESS],
-        );
-
-        Ok(SignedAggregateAndProof::from_aggregate_and_proof(
-            message, signature,
-        ))
-    }
-
     /// Produces a `SelectionProof` for the `slot`, signed by with corresponding secret key to
     /// `validator_pubkey`.
     async fn produce_selection_proof(
@@ -1090,80 +1178,172 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore for LighthouseValidatorS
         Ok(signature.into())
     }
 
-    async fn produce_sync_committee_signature(
+    fn sign_aggregate_and_proofs(
-        &self,
+        self: &Arc<Self>,
-        slot: Slot,
+        aggregates: Vec<AggregateToSign<E>>,
-        beacon_block_root: Hash256,
+    ) -> impl Stream<Item = Result<Vec<SignedAggregateAndProof<E>>, Error>> + Send {
-        validator_index: u64,
+        let store = self.clone();
-        validator_pubkey: &PublicKeyBytes,
+        let count = aggregates.len();
-    ) -> Result<SyncCommitteeMessage, Error> {
+        stream::once(async move {
-        let signing_epoch = slot.epoch(E::slots_per_epoch());
+            let signing_futures = aggregates.into_iter().map(
-        let signing_context = self.signing_context(Domain::SyncCommittee, signing_epoch);
+                |AggregateToSign {
-
+                     pubkey,
-        // Bypass `with_validator_signing_method`: sync committee messages are not slashable.
+                     aggregator_index,
-        let signing_method = self.doppelganger_bypassed_signing_method(*validator_pubkey)?;
+                     aggregate,
-
+                     selection_proof,
-        let signature = signing_method
+                 }| {
-            .get_signature::<E, BlindedPayload<E>>(
+                    let store = store.clone();
-                SignableMessage::SyncCommitteeSignature {
+                    async move {
-                    beacon_block_root,
+                        let result = store
-                    slot,
+                            .produce_signed_aggregate_and_proof(
-                },
+                                pubkey,
-                signing_context,
+                                aggregator_index,
-                &self.spec,
+                                aggregate,
-                &self.task_executor,
+                                selection_proof,
                             )
-            .await
+                            .await;
-            .map_err(Error::SpecificError)?;
+                        (pubkey, result)
-
+                    }
-        validator_metrics::inc_counter_vec(
+                },
-            &validator_metrics::SIGNED_SYNC_COMMITTEE_MESSAGES_TOTAL,
-            &[validator_metrics::SUCCESS],
             );
 
-        Ok(SyncCommitteeMessage {
+            let results = join_all(signing_futures)
-            slot,
+                .instrument(info_span!("sign_aggregates", count))
-            beacon_block_root,
+                .await;
-            validator_index,
+
-            signature,
+            let mut signed = Vec::with_capacity(results.len());
+            for (pubkey, result) in results {
+                match result {
+                    Ok(agg) => signed.push(agg),
+                    Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
+                        // A pubkey can be missing when a validator was recently
+                        // removed via the API.
+                        debug!(?pubkey, "Missing pubkey for aggregate");
+                    }
+                    Err(e) => {
+                        crit!(error = ?e, pubkey = ?pubkey, "Failed to sign aggregate");
+                    }
+                }
+            }
+            Ok(signed)
         })
     }
 
-    async fn produce_signed_contribution_and_proof(
+    fn sign_sync_committee_signatures(
-        &self,
+        self: &Arc<Self>,
-        aggregator_index: u64,
+        messages: Vec<SyncMessageToSign>,
-        aggregator_pubkey: PublicKeyBytes,
+    ) -> impl Stream<Item = Result<Vec<SyncCommitteeMessage>, Error>> + Send {
-        contribution: SyncCommitteeContribution<E>,
+        let store = self.clone();
-        selection_proof: SyncSelectionProof,
+        let count = messages.len();
-    ) -> Result<SignedContributionAndProof<E>, Error> {
+        stream::once(async move {
-        let signing_epoch = contribution.slot.epoch(E::slots_per_epoch());
+            let signing_futures = messages.into_iter().map(
-        let signing_context = self.signing_context(Domain::ContributionAndProof, signing_epoch);
+                |SyncMessageToSign {
-
+                     slot,
-        // Bypass `with_validator_signing_method`: sync committee messages are not slashable.
+                     beacon_block_root,
-        let signing_method = self.doppelganger_bypassed_signing_method(aggregator_pubkey)?;
+                     validator_index,
-
+                     pubkey,
-        let message = ContributionAndProof {
+                 }| {
-            aggregator_index,
+                    let store = store.clone();
-            contribution,
+                    async move {
-            selection_proof: selection_proof.into(),
+                        let result = store
-        };
+                            .produce_sync_committee_signature(
-
+                                slot,
-        let signature = signing_method
+                                beacon_block_root,
-            .get_signature::<E, BlindedPayload<E>>(
+                                validator_index,
-                SignableMessage::SignedContributionAndProof(&message),
+                                &pubkey,
-                signing_context,
-                &self.spec,
-                &self.task_executor,
                             )
-            .await
+                            .await;
-            .map_err(Error::SpecificError)?;
+                        (pubkey, validator_index, slot, result)
-
+                    }
-        validator_metrics::inc_counter_vec(
+                },
-            &validator_metrics::SIGNED_SYNC_COMMITTEE_CONTRIBUTIONS_TOTAL,
-            &[validator_metrics::SUCCESS],
             );
 
-        Ok(SignedContributionAndProof { message, signature })
+            let results = join_all(signing_futures)
+                .instrument(info_span!("sign_sync_signatures", count))
+                .await;
+
+            let mut signed = Vec::with_capacity(results.len());
+            for (_pubkey, validator_index, slot, result) in results {
+                match result {
+                    Ok(sig) => signed.push(sig),
+                    Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
+                        // A pubkey can be missing when a validator was recently
+                        // removed via the API.
+                        debug!(
+                            ?pubkey,
+                            validator_index,
+                            %slot,
+                            "Missing pubkey for sync committee signature"
+                        );
+                    }
+                    Err(e) => {
+                        crit!(
+                            validator_index,
+                            %slot,
+                            error = ?e,
+                            "Failed to sign sync committee signature"
+                        );
+                    }
+                }
+            }
+            Ok(signed)
+        })
+    }
+
+    fn sign_sync_committee_contributions(
+        self: &Arc<Self>,
+        contributions: Vec<ContributionToSign<E>>,
+    ) -> impl Stream<Item = Result<Vec<SignedContributionAndProof<E>>, Error>> + Send {
+        let store = self.clone();
+        let count = contributions.len();
+        stream::once(async move {
+            let signing_futures = contributions.into_iter().map(
+                |ContributionToSign {
+                     aggregator_index,
+                     aggregator_pubkey,
+                     contribution,
+                     selection_proof,
+                 }| {
+                    let store = store.clone();
+                    let slot = contribution.slot;
+                    async move {
+                        let result = store
+                            .produce_signed_contribution_and_proof(
+                                aggregator_index,
+                                aggregator_pubkey,
+                                contribution,
+                                selection_proof,
+                            )
+                            .await;
+                        (slot, result)
+                    }
+                },
+            );
+
+            let results = join_all(signing_futures)
+                .instrument(info_span!("sign_sync_contributions", count))
+                .await;
+
+            let mut signed = Vec::with_capacity(results.len());
+            for (slot, result) in results {
+                match result {
+                    Ok(contribution) => signed.push(contribution),
+                    Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
+                        // A pubkey can be missing when a validator was recently
+                        // removed via the API.
+                        debug!(?pubkey, %slot, "Missing pubkey for sync contribution");
+                    }
+                    Err(e) => {
+                        crit!(
+                            %slot,
+                            error = ?e,
+                            "Unable to sign sync committee contribution"
+                        );
+                    }
+                }
+            }
+            Ok(signed)
+        })
     }
 
     /// Prune the slashing protection database so that it remains performant.

validator_client/validator_services/src/attestation_service.rs

@@ -1,6 +1,6 @@
 use crate::duties_service::{DutiesService, DutyAndProof};
 use beacon_node_fallback::{ApiTopic, BeaconNodeFallback, beacon_head_monitor::HeadEvent};
-use futures::future::join_all;
+use futures::StreamExt;
 use logging::crit;
 use slot_clock::SlotClock;
 use std::collections::HashMap;
@@ -13,7 +13,7 @@ use tokio::time::{Duration, Instant, sleep, sleep_until};
 use tracing::{Instrument, debug, error, info, info_span, instrument, warn};
 use tree_hash::TreeHash;
 use types::{Attestation, AttestationData, ChainSpec, CommitteeIndex, EthSpec, Hash256, Slot};
-use validator_store::{Error as ValidatorStoreError, ValidatorStore};
+use validator_store::{AggregateToSign, AttestationToSign, ValidatorStore};
 
 /// Builds an `AttestationService`.
 #[derive(Default)]
@@ -560,12 +560,12 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                 }
             };
 
-            attestations_to_sign.push((
+            attestations_to_sign.push(AttestationToSign {
-                duty.validator_index,
+                validator_index: duty.validator_index,
-                duty.pubkey,
+                pubkey: duty.pubkey,
-                duty.validator_committee_index as usize,
+                validator_committee_index: duty.validator_committee_index as usize,
                 attestation,
-            ));
+            });
         }
 
         if attestations_to_sign.is_empty() {
@@ -573,26 +573,27 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             return Ok(());
         }
 
-        // Sign and check all attestations (includes slashing protection).
+        let attestation_stream = self.validator_store.sign_attestations(attestations_to_sign);
-        let safe_attestations = self
+        tokio::pin!(attestation_stream);
-            .validator_store
-            .sign_attestations(attestations_to_sign)
-            .await
-            .map_err(|e| format!("Failed to sign attestations: {e:?}"))?;
 
-        if safe_attestations.is_empty() {
-            warn!("No attestations were published");
-            return Ok(());
-        }
         let fork_name = self
             .chain_spec
             .fork_name_at_slot::<S::E>(attestation_data.slot);
 
-        let single_attestations = safe_attestations
+        // Publish each batch as it arrives from the stream.
+        let mut received_non_empty_batch = false;
+        while let Some(result) = attestation_stream.next().await {
+            match result {
+                Ok(batch) if !batch.is_empty() => {
+                    received_non_empty_batch = true;
+
+                    let single_attestations = batch
                         .iter()
-            .filter_map(|(i, a)| {
+                        .filter_map(|(attester_index, attestation)| {
-                match a.to_single_attestation_with_attester_index(*i) {
+                            match attestation
-                    Ok(a) => Some(a),
+                                .to_single_attestation_with_attester_index(*attester_index)
+                            {
+                                Ok(single_attestation) => Some(single_attestation),
                                 Err(e) => {
                                     // This shouldn't happen unless BN and VC are out of sync with
                                     // respect to the Electra fork.
@@ -651,6 +652,17 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             "Unable to publish attestations"
                         ),
                     }
+                }
+                Err(e) => {
+                    crit!(error = ?e, "Failed to sign attestations");
+                }
+                _ => {}
+            }
+        }
+
+        if !received_non_empty_batch {
+            warn!("No attestations were published");
+        }
 
         Ok(())
     }
@@ -725,8 +737,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
             .await
             .map_err(|e| e.to_string())?;
 
-        // Create futures to produce the signed aggregated attestations.
+        // Build the batch of aggregates to sign.
-        let signing_futures = validator_duties.iter().map(|duty_and_proof| async move {
+        let aggregates_to_sign: Vec<_> = validator_duties
+            .iter()
+            .filter_map(|duty_and_proof| {
                 let duty = &duty_and_proof.duty;
                 let selection_proof = duty_and_proof.selection_proof.as_ref()?;
 
@@ -735,48 +749,26 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     return None;
                 }
 
-            match self
+                Some(AggregateToSign {
+                    pubkey: duty.pubkey,
+                    aggregator_index: duty.validator_index,
+                    aggregate: aggregated_attestation.clone(),
+                    selection_proof: selection_proof.clone(),
+                })
+            })
+            .collect();
+
+        // Sign aggregates. Returns a stream of batches.
+        let aggregate_stream = self
             .validator_store
-                .produce_signed_aggregate_and_proof(
+            .sign_aggregate_and_proofs(aggregates_to_sign);
-                    duty.pubkey,
+        tokio::pin!(aggregate_stream);
-                    duty.validator_index,
-                    aggregated_attestation.clone(),
-                    selection_proof.clone(),
-                )
-                .await
-            {
-                Ok(aggregate) => Some(aggregate),
-                Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
-                    // A pubkey can be missing when a validator was recently
-                    // removed via the API.
-                    debug!(?pubkey, "Missing pubkey for aggregate");
-                    None
-                }
-                Err(e) => {
-                    crit!(
-                        error = ?e,
-                        pubkey = ?duty.pubkey,
-                        "Failed to sign aggregate"
-                    );
-                    None
-                }
-            }
-        });
 
-        // Execute all the futures in parallel, collecting any successful results.
+        // Publish each batch as it arrives from the stream.
-        let aggregator_count = validator_duties
+        while let Some(result) = aggregate_stream.next().await {
-            .iter()
+            match result {
-            .filter(|d| d.selection_proof.is_some())
+                Ok(batch) if !batch.is_empty() => {
-            .count();
+                    let signed_aggregate_and_proofs = batch.as_slice();
-        let signed_aggregate_and_proofs = join_all(signing_futures)
-            .instrument(info_span!("sign_aggregates", count = aggregator_count))
-            .await
-            .into_iter()
-            .flatten()
-            .collect::<Vec<_>>();
-
-        if !signed_aggregate_and_proofs.is_empty() {
-            let signed_aggregate_and_proofs_slice = signed_aggregate_and_proofs.as_slice();
                     match self
                         .beacon_nodes
                         .first_success(|beacon_node| async move {
@@ -787,14 +779,14 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             if fork_name.electra_enabled() {
                                 beacon_node
                                     .post_validator_aggregate_and_proof_v2(
-                                signed_aggregate_and_proofs_slice,
+                                        signed_aggregate_and_proofs,
                                         fork_name,
                                     )
                                     .await
                             } else {
                                 beacon_node
                                     .post_validator_aggregate_and_proof_v1(
-                                signed_aggregate_and_proofs_slice,
+                                        signed_aggregate_and_proofs,
                                     )
                                     .await
                             }
@@ -809,9 +801,11 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             for signed_aggregate_and_proof in signed_aggregate_and_proofs {
                                 let attestation = signed_aggregate_and_proof.message().aggregate();
                                 info!(
-                            aggregator = signed_aggregate_and_proof.message().aggregator_index(),
+                                    aggregator =
+                                        signed_aggregate_and_proof.message().aggregator_index(),
                                     signatures = attestation.num_set_aggregation_bits(),
-                            head_block = format!("{:?}", attestation.data().beacon_block_root),
+                                    head_block =
+                                        format!("{:?}", attestation.data().beacon_block_root),
                                     committee_index = attestation.committee_index(),
                                     slot = attestation.data().slot.as_u64(),
                                     "type" = "aggregated",
@@ -824,7 +818,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                                 let attestation = &signed_aggregate_and_proof.message().aggregate();
                                 crit!(
                                     error = %e,
-                            aggregator = signed_aggregate_and_proof.message().aggregator_index(),
+                                    aggregator = signed_aggregate_and_proof
+                                        .message()
+                                        .aggregator_index(),
                                     committee_index = attestation.committee_index(),
                                     slot = attestation.data().slot.as_u64(),
                                     "type" = "aggregated",
@@ -834,6 +830,12 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                         }
                     }
                 }
+                Err(e) => {
+                    crit!(error = ?e, "Failed to sign aggregates");
+                }
+                _ => {}
+            }
+        }
 
         Ok(())
     }

validator_client/validator_services/src/sync_committee_service.rs

@@ -2,8 +2,8 @@ use crate::duties_service::DutiesService;
 use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
 use bls::PublicKeyBytes;
 use eth2::types::BlockId;
+use futures::StreamExt;
 use futures::future::FutureExt;
-use futures::future::join_all;
 use logging::crit;
 use slot_clock::SlotClock;
 use std::collections::HashMap;
@@ -17,7 +17,7 @@ use types::{
     ChainSpec, EthSpec, Hash256, Slot, SyncCommitteeSubscription, SyncContributionData, SyncDuty,
     SyncSelectionProof, SyncSubnetId,
 };
-use validator_store::{Error as ValidatorStoreError, ValidatorStore};
+use validator_store::{ContributionToSign, SyncMessageToSign, ValidatorStore};
 
 pub const SUBSCRIPTION_LOOKAHEAD_EPOCHS: u64 = 4;
 
@@ -247,54 +247,27 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
         beacon_block_root: Hash256,
         validator_duties: Vec<SyncDuty>,
     ) -> Result<(), ()> {
-        // Create futures to produce sync committee signatures.
+        let messages_to_sign: Vec<_> = validator_duties
-        let signature_futures = validator_duties.iter().map(|duty| async move {
+            .iter()
-            match self
+            .map(|duty| SyncMessageToSign {
-                .validator_store
-                .produce_sync_committee_signature(
                 slot,
                 beacon_block_root,
-                    duty.validator_index,
+                validator_index: duty.validator_index,
-                    &duty.pubkey,
+                pubkey: duty.pubkey,
-                )
+            })
-                .await
+            .collect();
-            {
-                Ok(signature) => Some(signature),
-                Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
-                    // A pubkey can be missing when a validator was recently
-                    // removed via the API.
-                    debug!(
-                        ?pubkey,
-                        validator_index = duty.validator_index,
-                        %slot,
-                        "Missing pubkey for sync committee signature"
-                    );
-                    None
-                }
-                Err(e) => {
-                    crit!(
-                        validator_index = duty.validator_index,
-                        %slot,
-                        error = ?e,
-                        "Failed to sign sync committee signature"
-                    );
-                    None
-                }
-            }
-        });
 
-        // Execute all the futures in parallel, collecting any successful results.
+        let signature_stream = self
-        let committee_signatures = &join_all(signature_futures)
+            .validator_store
-            .instrument(info_span!(
+            .sign_sync_committee_signatures(messages_to_sign);
-                "sign_sync_signatures",
+        tokio::pin!(signature_stream);
-                count = validator_duties.len()
-            ))
-            .await
-            .into_iter()
-            .flatten()
-            .collect::<Vec<_>>();
 
-        self.beacon_nodes
+        while let Some(result) = signature_stream.next().await {
+            match result {
+                Ok(committee_signatures) if !committee_signatures.is_empty() => {
+                    let committee_signatures = &committee_signatures;
+                    match self
+                        .beacon_nodes
                         .request(ApiTopic::SyncCommittee, |beacon_node| async move {
                             beacon_node
                                 .post_beacon_pool_sync_committee_signatures(committee_signatures)
@@ -305,20 +278,26 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                             count = committee_signatures.len()
                         ))
                         .await
-            .map_err(|e| {
+                    {
-                error!(
+                        Ok(()) => info!(
-                    %slot,
-                    error = %e,
-                    "Unable to publish sync committee messages"
-                );
-            })?;
-
-        info!(
                             count = committee_signatures.len(),
                             head_block = ?beacon_block_root,
                             %slot,
                             "Successfully published sync committee messages"
-        );
+                        ),
+                        Err(e) => error!(
+                            %slot,
+                            error = %e,
+                            "Unable to publish sync committee messages"
+                        ),
+                    }
+                }
+                Err(e) => {
+                    crit!(%slot, error = ?e, "Failed to sign sync committee signatures");
+                }
+                _ => {}
+            }
+        }
 
         Ok(())
     }
@@ -389,52 +368,30 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
             })?
             .data;
 
-        // Create futures to produce signed contributions.
+        let contributions_to_sign: Vec<_> = subnet_aggregators
-        let aggregator_count = subnet_aggregators.len();
-        let signature_futures = subnet_aggregators.into_iter().map(
-            |(aggregator_index, aggregator_pk, selection_proof)| async move {
-                match self
-                    .validator_store
-                    .produce_signed_contribution_and_proof(
-                        aggregator_index,
-                        aggregator_pk,
-                        contribution.clone(),
-                        selection_proof,
-                    )
-                    .await
-                {
-                    Ok(signed_contribution) => Some(signed_contribution),
-                    Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
-                        // A pubkey can be missing when a validator was recently
-                        // removed via the API.
-                        debug!(?pubkey, %slot, "Missing pubkey for sync contribution");
-                        None
-                    }
-                    Err(e) => {
-                        crit!(
-                            %slot,
-                            error = ?e,
-                            "Unable to sign sync committee contribution"
-                        );
-                        None
-                    }
-                }
-            },
-        );
-
-        // Execute all the futures in parallel, collecting any successful results.
-        let signed_contributions = &join_all(signature_futures)
-            .instrument(info_span!(
-                "sign_sync_contributions",
-                count = aggregator_count
-            ))
-            .await
             .into_iter()
-            .flatten()
+            .map(
-            .collect::<Vec<_>>();
+                |(aggregator_index, aggregator_pk, selection_proof)| ContributionToSign {
+                    aggregator_index,
+                    aggregator_pubkey: aggregator_pk,
+                    contribution: contribution.clone(),
+                    selection_proof,
+                },
+            )
+            .collect();
 
+        let contribution_stream = self
+            .validator_store
+            .sign_sync_committee_contributions(contributions_to_sign);
+        tokio::pin!(contribution_stream);
+
+        while let Some(result) = contribution_stream.next().await {
+            match result {
+                Ok(signed_contributions) if !signed_contributions.is_empty() => {
+                    let signed_contributions = &signed_contributions;
                     // Publish to the beacon node.
-        self.beacon_nodes
+                    match self
+                        .beacon_nodes
                         .first_success(|beacon_node| async move {
                             beacon_node
                                 .post_validator_contribution_and_proofs(signed_contributions)
@@ -445,21 +402,27 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                             count = signed_contributions.len()
                         ))
                         .await
-            .map_err(|e| {
+                    {
-                error!(
+                        Ok(()) => info!(
-                    %slot,
-                    error = %e,
-                    "Unable to publish signed contributions and proofs"
-                );
-            })?;
-
-        info!(
                             subnet = %subnet_id,
                             beacon_block_root = %beacon_block_root,
                             num_signers = contribution.aggregation_bits.num_set_bits(),
                             %slot,
                             "Successfully published sync contributions"
-        );
+                        ),
+                        Err(e) => error!(
+                            %slot,
+                            error = %e,
+                            "Unable to publish signed contributions and proofs"
+                        ),
+                    }
+                }
+                Err(e) => {
+                    crit!(%slot, error = ?e, "Failed to sign sync committee contributions");
+                }
+                _ => {}
+            }
+        }
 
         Ok(())
     }

validator_client/validator_store/Cargo.toml

@@ -7,5 +7,6 @@ authors = ["Sigma Prime <contact@sigmaprime.io>"]
 [dependencies]
 bls = { workspace = true }
 eth2 = { workspace = true }
+futures = { workspace = true }
 slashing_protection = { workspace = true }
 types = { workspace = true }

validator_client/validator_store/src/lib.rs

@@ -1,5 +1,6 @@
 use bls::{PublicKeyBytes, Signature};
 use eth2::types::{FullBlockContents, PublishBlockRequest};
+use futures::Stream;
 use slashing_protection::NotSafe;
 use std::fmt::Debug;
 use std::future::Future;
@@ -32,6 +33,38 @@ impl<T> From<T> for Error<T> {
     }
 }
 
+/// Input for batch attestation signing
+pub struct AttestationToSign<E: EthSpec> {
+    pub validator_index: u64,
+    pub pubkey: PublicKeyBytes,
+    pub validator_committee_index: usize,
+    pub attestation: Attestation<E>,
+}
+
+/// Input for batch aggregate signing
+pub struct AggregateToSign<E: EthSpec> {
+    pub pubkey: PublicKeyBytes,
+    pub aggregator_index: u64,
+    pub aggregate: Attestation<E>,
+    pub selection_proof: SelectionProof,
+}
+
+/// Input for batch sync committee message signing
+pub struct SyncMessageToSign {
+    pub slot: Slot,
+    pub beacon_block_root: Hash256,
+    pub validator_index: u64,
+    pub pubkey: PublicKeyBytes,
+}
+
+/// Input for batch sync committee contribution signing
+pub struct ContributionToSign<E: EthSpec> {
+    pub aggregator_index: u64,
+    pub aggregator_pubkey: PublicKeyBytes,
+    pub contribution: SyncCommitteeContribution<E>,
+    pub selection_proof: SyncSelectionProof,
+}
+
 /// A helper struct, used for passing data from the validator store to services.
 pub struct ProposalData {
     pub validator_index: Option<u64>,
@@ -106,13 +139,9 @@ pub trait ValidatorStore: Send + Sync {
 
     /// Sign a batch of `attestations` and apply slashing protection to them.
     ///
-    /// Only successfully signed attestations that pass slashing protection are returned, along with
+    /// Returns a stream of batches of successfully signed attestations. Each batch contains
-    /// the validator index of the signer. Eventually this will be replaced by `SingleAttestation`
+    /// attestations that passed slashing protection, along with the validator index of the signer.
-    /// use.
+    /// Eventually this will be replaced by `SingleAttestation` use.
-    ///
-    /// Input:
-    ///
-    /// * Vec of (validator_index, pubkey, validator_committee_index, attestation).
     ///
     /// Output:
     ///
@@ -120,26 +149,14 @@ pub trait ValidatorStore: Send + Sync {
     #[allow(clippy::type_complexity)]
     fn sign_attestations(
         self: &Arc<Self>,
-        attestations: Vec<(u64, PublicKeyBytes, usize, Attestation<Self::E>)>,
+        attestations: Vec<AttestationToSign<Self::E>>,
-    ) -> impl Future<Output = Result<Vec<(u64, Attestation<Self::E>)>, Error<Self::Error>>> + Send;
+    ) -> impl Stream<Item = Result<Vec<(u64, Attestation<Self::E>)>, Error<Self::Error>>> + Send;
 
     fn sign_validator_registration_data(
         &self,
         validator_registration_data: ValidatorRegistrationData,
     ) -> impl Future<Output = Result<SignedValidatorRegistrationData, Error<Self::Error>>> + Send;
 
-    /// Signs an `AggregateAndProof` for a given validator.
-    ///
-    /// The resulting `SignedAggregateAndProof` is sent on the aggregation channel and cannot be
-    /// modified by actors other than the signing validator.
-    fn produce_signed_aggregate_and_proof(
-        &self,
-        validator_pubkey: PublicKeyBytes,
-        aggregator_index: u64,
-        aggregate: Attestation<Self::E>,
-        selection_proof: SelectionProof,
-    ) -> impl Future<Output = Result<SignedAggregateAndProof<Self::E>, Error<Self::Error>>> + Send;
-
     /// Produces a `SelectionProof` for the `slot`, signed by with corresponding secret key to
     /// `validator_pubkey`.
     fn produce_selection_proof(
@@ -156,21 +173,23 @@ pub trait ValidatorStore: Send + Sync {
         subnet_id: SyncSubnetId,
     ) -> impl Future<Output = Result<SyncSelectionProof, Error<Self::Error>>> + Send;
 
-    fn produce_sync_committee_signature(
+    /// Sign a batch of aggregate and proofs and return results as a stream of batches.
-        &self,
+    fn sign_aggregate_and_proofs(
-        slot: Slot,
+        self: &Arc<Self>,
-        beacon_block_root: Hash256,
+        aggregates: Vec<AggregateToSign<Self::E>>,
-        validator_index: u64,
+    ) -> impl Stream<Item = Result<Vec<SignedAggregateAndProof<Self::E>>, Error<Self::Error>>> + Send;
-        validator_pubkey: &PublicKeyBytes,
-    ) -> impl Future<Output = Result<SyncCommitteeMessage, Error<Self::Error>>> + Send;
 
-    fn produce_signed_contribution_and_proof(
+    /// Sign a batch of sync committee messages and return results as a stream of batches.
-        &self,
+    fn sign_sync_committee_signatures(
-        aggregator_index: u64,
+        self: &Arc<Self>,
-        aggregator_pubkey: PublicKeyBytes,
+        messages: Vec<SyncMessageToSign>,
-        contribution: SyncCommitteeContribution<Self::E>,
+    ) -> impl Stream<Item = Result<Vec<SyncCommitteeMessage>, Error<Self::Error>>> + Send;
-        selection_proof: SyncSelectionProof,
+
-    ) -> impl Future<Output = Result<SignedContributionAndProof<Self::E>, Error<Self::Error>>> + Send;
+    /// Sign a batch of sync committee contributions and return results as a stream of batches.
+    fn sign_sync_committee_contributions(
+        self: &Arc<Self>,
+        contributions: Vec<ContributionToSign<Self::E>>,
+    ) -> impl Stream<Item = Result<Vec<SignedContributionAndProof<Self::E>>, Error<Self::Error>>> + Send;
 
     /// Prune the slashing protection database so that it remains performant.
     ///