Update /validator/subscribe (#969)

* Add progress on duties refactor

* Add simple is_aggregator bool to val subscription

* Remove unused function

validator_client/src/attestation_service.rs

@@ -256,48 +256,34 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
     /// slot allowing the beacon node to connect to the required subnet and determine
     /// if attestations need to be aggregated.
     fn send_subscriptions(&self, duties: Vec<DutyAndState>) -> impl Future<Item = (), Error = ()> {
-        let mut validator_subscriptions = Vec::new();
-        let mut successful_duties = Vec::new();
-
         let service_1 = self.clone();
-        let duties_no = duties.len();
+        let num_duties = duties.len();
 
         let log_1 = self.context.log.clone();
         let log_2 = self.context.log.clone();
 
-        // builds a list of subscriptions
-        for duty in duties {
-            if let Some((slot, attestation_committee_index, _, validator_index)) =
-                duty.attestation_duties()
-            {
-                if let Some(slot_signature) = self
+        let (validator_subscriptions, successful_duties): (Vec<_>, Vec<_>) = duties
+            .into_iter()
+            .filter_map(|duty| {
+                let (slot, attestation_committee_index, _, validator_index) =
+                    duty.attestation_duties()?;
+                let selection_proof = self
                     .validator_store
-                    .sign_slot(duty.validator_pubkey(), slot)
-                {
-                    let is_aggregator_proof = if duty.is_aggregator() {
-                        Some(slot_signature.clone())
-                    } else {
-                        None
-                    };
+                    .produce_selection_proof(duty.validator_pubkey(), slot)?;
+                let modulo = duty.duty.aggregator_modulo?;
 
-                    let subscription = ValidatorSubscription::new(
-                        validator_index,
-                        attestation_committee_index,
-                        slot,
-                        slot_signature,
-                    );
-                    validator_subscriptions.push(subscription);
+                let subscription = ValidatorSubscription {
+                    validator_index,
+                    attestation_committee_index,
+                    slot,
+                    is_aggregator: selection_proof.is_aggregator(modulo),
+                };
 
-                    // add successful duties to the list, along with whether they are aggregation
-                    // duties or not
-                    successful_duties.push((duty, is_aggregator_proof));
-                }
-            } else {
-                crit!(log_2, "Validator duty doesn't have required fields");
-            }
-        }
+                Some((subscription, (duty, selection_proof)))
+            })
+            .unzip();
 
-        let failed_duties = duties_no - successful_duties.len();
+        let num_failed_duties = num_duties - successful_duties.len();
 
         self.beacon_node
             .http
@@ -308,8 +294,8 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
                 PublishStatus::Valid => info!(
                     log_1,
                     "Successfully subscribed validators";
-                    "validators" => duties_no,
-                    "failed_validators" => failed_duties,
+                    "validators" => num_duties,
+                    "failed_validators" => num_failed_duties,
                 ),
                 PublishStatus::Invalid(msg) => crit!(
                     log_1,
@@ -321,10 +307,10 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
                 }
             })
             .and_then(move |_| {
-                for (duty, is_aggregator_proof) in successful_duties {
+                for (duty, selection_proof) in successful_duties {
                     service_1
                         .duties_service
-                        .subscribe_duty(&duty.duty, is_aggregator_proof);
+                        .subscribe_duty(&duty.duty, selection_proof);
                 }
                 Ok(())
             })

validator_client/src/duties_service.rs

@@ -1,5 +1,4 @@
 use crate::validator_store::ValidatorStore;
-use bls::Signature;
 use environment::RuntimeContext;
 use exit_future::Signal;
 use futures::{future, Future, IntoFuture, Stream};
@@ -14,7 +13,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::timer::Interval;
-use types::{ChainSpec, CommitteeIndex, Epoch, EthSpec, PublicKey, Slot};
+use types::{ChainSpec, CommitteeIndex, Epoch, EthSpec, PublicKey, SelectionProof, Slot};
 
 /// Delay this period of time after the slot starts. This allows the node to process the new slot.
 const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
@@ -28,11 +27,9 @@ type BaseHashMap = HashMap<PublicKey, HashMap<Epoch, DutyAndState>>;
 pub enum DutyState {
     /// This duty has not been subscribed to the beacon node.
     NotSubscribed,
-    /// The duty has been subscribed to the beacon node.
-    Subscribed,
     /// The duty has been subscribed and the validator is an aggregator for this duty. The
     /// selection proof is provided to construct the `AggregateAndProof` struct.
-    SubscribedAggregator(Signature),
+    SubscribedAggregator(SelectionProof),
 }
 
 #[derive(Debug, Clone)]
@@ -49,13 +46,12 @@ impl DutyAndState {
     pub fn is_aggregator(&self) -> bool {
         match self.state {
             DutyState::NotSubscribed => false,
-            DutyState::Subscribed => false,
             DutyState::SubscribedAggregator(_) => true,
         }
     }
 
     /// Returns the selection proof if the duty is an aggregation duty.
-    pub fn selection_proof(&self) -> Option<Signature> {
+    pub fn selection_proof(&self) -> Option<SelectionProof> {
         match &self.state {
             DutyState::SubscribedAggregator(proof) => Some(proof.clone()),
             _ => None,
@@ -66,7 +62,6 @@ impl DutyAndState {
     pub fn is_subscribed(&self) -> bool {
         match self.state {
             DutyState::NotSubscribed => false,
-            DutyState::Subscribed => true,
             DutyState::SubscribedAggregator(_) => true,
         }
     }
@@ -425,14 +420,14 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
     /// Marks the duty as being subscribed to the beacon node.
     ///
     /// If the duty is to be marked as an aggregator duty, a selection proof is also provided.
-    pub fn subscribe_duty(&self, duty: &ValidatorDuty, aggregator_proof: Option<Signature>) {
-        let state = match aggregator_proof {
-            Some(proof) => DutyState::SubscribedAggregator(proof),
-            None => DutyState::Subscribed,
-        };
+    pub fn subscribe_duty(&self, duty: &ValidatorDuty, proof: SelectionProof) {
         if let Some(slot) = duty.attestation_slot {
-            self.store
-                .set_duty_state(&duty.validator_pubkey, slot, state, E::slots_per_epoch())
+            self.store.set_duty_state(
+                &duty.validator_pubkey,
+                slot,
+                DutyState::SubscribedAggregator(proof),
+                E::slots_per_epoch(),
+            )
         }
     }
 

validator_client/src/validator_store.rs

@@ -12,8 +12,8 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use tempdir::TempDir;
 use types::{
-    Attestation, BeaconBlock, ChainSpec, Domain, Epoch, EthSpec, Fork, PublicKey, Signature,
-    SignedAggregateAndProof, SignedBeaconBlock, SignedRoot, Slot,
+    Attestation, BeaconBlock, ChainSpec, Domain, Epoch, EthSpec, Fork, PublicKey, SelectionProof,
+    Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedRoot, Slot,
 };
 
 #[derive(Clone)]
@@ -221,22 +221,21 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         ))
     }
 
-    /// Signs a slot for a given validator.
-    ///
-    /// This is used to subscribe a validator to a beacon node and is used to determine if the
-    /// validator is to aggregate attestations for this slot.
-    pub fn sign_slot(&self, validator_pubkey: &PublicKey, slot: Slot) -> Option<Signature> {
+    /// Produces a `SelectionProof` for the `slot`, signed by with corresponding secret key to
+    /// `validator_pubkey`.
+    pub fn produce_selection_proof(
+        &self,
+        validator_pubkey: &PublicKey,
+        slot: Slot,
+    ) -> Option<SelectionProof> {
         let validators = self.validators.read();
         let voting_keypair = validators.get(validator_pubkey)?.voting_keypair.as_ref()?;
 
-        let domain = self.spec.get_domain(
-            slot.epoch(E::slots_per_epoch()),
-            Domain::SelectionProof,
+        Some(SelectionProof::new::<E>(
+            slot,
+            &voting_keypair.sk,
             &self.fork()?,
-        );
-
-        let message = slot.signing_root(domain);
-
-        Some(Signature::new(message.as_bytes(), &voting_keypair.sk))
+            &self.spec,
+        ))
     }
 }