Merge @thojest validator client to branch.

validator_client/src/attester_service/attestation_grpc_client.rs

@@ -0,0 +1,44 @@
+use protos::services_grpc::AttestationServiceClient;
+use std::sync::Arc;
+
+use attester::{BeaconNode, BeaconNodeError, PublishOutcome};
+use protos::services::ProduceAttestationDataRequest;
+use types::{AttestationData, FreeAttestation, Slot};
+
+pub struct AttestationGrpcClient {
+    client: Arc<AttestationServiceClient>,
+}
+
+impl AttestationGrpcClient {
+    pub fn new(client: Arc<AttestationServiceClient>) -> Self {
+        Self { client }
+    }
+}
+
+impl BeaconNode for AttestationGrpcClient {
+    fn produce_attestation_data(
+        &self,
+        slot: Slot,
+        shard: u64,
+    ) -> Result<Option<AttestationData>, BeaconNodeError> {
+        let mut req = ProduceAttestationDataRequest::new();
+        req.set_slot(slot.as_u64());
+        req.set_shard(shard);
+
+        let reply = self
+            .client
+            .produce_attestation_data(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        // TODO: return correct AttestationData
+        Err(BeaconNodeError::DecodeFailure)
+    }
+
+    fn publish_attestation_data(
+        &self,
+        free_attestation: FreeAttestation,
+    ) -> Result<PublishOutcome, BeaconNodeError> {
+        // TODO: return correct PublishOutcome
+        Err(BeaconNodeError::DecodeFailure)
+    }
+}

validator_client/src/attester_service/mod.rs

@@ -0,0 +1,54 @@
+mod attestation_grpc_client;
+use attester::{Attester, BeaconNode, DutiesReader, PollOutcome as AttesterPollOutcome, Signer};
+use slog::{error, info, warn, Logger};
+use slot_clock::SlotClock;
+use std::time::Duration;
+
+pub use self::attestation_grpc_client::AttestationGrpcClient;
+
+pub struct AttesterService<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> {
+    pub attester: Attester<T, U, V, W>,
+    pub poll_interval_millis: u64,
+    pub log: Logger,
+}
+
+impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> AttesterService<T, U, V, W> {
+    /// Run a loop which polls the Attester each `poll_interval_millis` millseconds.
+    ///
+    /// Logs the results of the polls.
+    pub fn run(&mut self) {
+        loop {
+            match self.attester.poll() {
+                Err(error) => {
+                    error!(self.log, "Attester poll error"; "error" => format!("{:?}", error))
+                }
+                Ok(AttesterPollOutcome::AttestationProduced(slot)) => {
+                    info!(self.log, "Produced Attestation"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::SlashableAttestationNotProduced(slot)) => {
+                    warn!(self.log, "Slashable attestation was not produced"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::AttestationNotRequired(slot)) => {
+                    info!(self.log, "Attestation not required"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::ProducerDutiesUnknown(slot)) => {
+                    error!(self.log, "Attestation duties unknown"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::SlotAlreadyProcessed(slot)) => {
+                    warn!(self.log, "Attempted to re-process slot"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::BeaconNodeUnableToProduceAttestation(slot)) => {
+                    error!(self.log, "Beacon node unable to produce attestation"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::SignerRejection(slot)) => {
+                    error!(self.log, "The cryptographic signer refused to sign the attestation"; "slot" => slot)
+                }
+                Ok(AttesterPollOutcome::ValidatorIsUnknown(slot)) => {
+                    error!(self.log, "The Beacon Node does not recognise the validator"; "slot" => slot)
+                }
+            };
+
+            std::thread::sleep(Duration::from_millis(self.poll_interval_millis));
+        }
+    }
+}

validator_client/src/main.rs

@@ -1,11 +1,16 @@
 use self::block_producer_service::{BeaconBlockGrpcClient, BlockProducerService};
 use self::duties::{DutiesManager, DutiesManagerService, EpochDutiesMap};
+use crate::attester_service::{AttestationGrpcClient, AttesterService};
 use crate::config::ClientConfig;
-use block_proposer::{test_utils::LocalSigner, BlockProducer};
+use attester::test_utils::EpochMap;
+use attester::{test_utils::LocalSigner as AttesterLocalSigner, Attester};
+use block_proposer::{test_utils::LocalSigner as BlockProposerLocalSigner, BlockProducer};
 use bls::Keypair;
 use clap::{App, Arg};
 use grpcio::{ChannelBuilder, EnvBuilder};
-use protos::services_grpc::{BeaconBlockServiceClient, ValidatorServiceClient};
+use protos::services_grpc::{
+    AttestationServiceClient, BeaconBlockServiceClient, ValidatorServiceClient,
+};
 use slog::{error, info, o, Drain};
 use slot_clock::SystemTimeSlotClock;
 use std::path::PathBuf;
@@ -13,6 +18,7 @@ use std::sync::Arc;
 use std::thread;
 use types::ChainSpec;
 
+mod attester_service;
 mod block_producer_service;
 mod config;
 mod duties;
@@ -102,6 +108,13 @@ fn main() {
         Arc::new(ValidatorServiceClient::new(ch))
     };
 
+    //Beacon node gRPC attester endpoints.
+    let attester_grpc_client = {
+        let env = Arc::new(EnvBuilder::new().build());
+        let ch = ChannelBuilder::new(env).connect(&config.server);
+        Arc::new(AttestationServiceClient::new(ch))
+    };
+
     // Spec
     let spec = Arc::new(config.spec.clone());
 
@@ -130,6 +143,7 @@ fn main() {
     for keypair in keypairs {
         info!(log, "Starting validator services"; "validator" => keypair.pk.concatenated_hex_id());
         let duties_map = Arc::new(EpochDutiesMap::new(spec.slots_per_epoch));
+        let epoch_map_for_attester = Arc::new(EpochMap::new(spec.slots_per_epoch));
 
         // Spawn a new thread to maintain the validator's `EpochDuties`.
         let duties_manager_thread = {
@@ -160,7 +174,7 @@ fn main() {
         // Spawn a new thread to perform block production for the validator.
         let producer_thread = {
             let spec = spec.clone();
-            let signer = Arc::new(LocalSigner::new(keypair.clone()));
+            let signer = Arc::new(BlockProposerLocalSigner::new(keypair.clone()));
             let duties_map = duties_map.clone();
             let slot_clock = slot_clock.clone();
             let log = log.clone();
@@ -178,13 +192,33 @@ fn main() {
             })
         };
 
-        threads.push((duties_manager_thread, producer_thread));
+        // Spawn a new thread for attestation for the validator.
+        let attester_thread = {
+            let signer = Arc::new(AttesterLocalSigner::new(keypair.clone()));
+            let epoch_map = epoch_map_for_attester.clone();
+            let slot_clock = slot_clock.clone();
+            let log = log.clone();
+            let client = Arc::new(AttestationGrpcClient::new(attester_grpc_client.clone()));
+            thread::spawn(move || {
+                let attester = Attester::new(epoch_map, slot_clock, client, signer);
+                let mut attester_service = AttesterService {
+                    attester,
+                    poll_interval_millis,
+                    log,
+                };
+
+                attester_service.run();
+            })
+        };
+
+        threads.push((duties_manager_thread, producer_thread, attester_thread));
     }
 
     // Naively wait for all the threads to complete.
     for tuple in threads {
-        let (manager, producer) = tuple;
+        let (manager, producer, attester) = tuple;
         let _ = producer.join();
         let _ = manager.join();
+        let _ = attester.join();
     }
 }