Add validator_node, restructure binaries, gRPC.

This is a massive commit which restructures the workspace, adds a very
basic, untested, validator client and some very basic, non-functioning
gRPC endpoints to the beacon-node.

validator_client/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "validator_client"
+version = "0.1.0"
+authors = ["Paul Hauner <paul@paulhauner.com>"]
+edition = "2018"
+
+[dependencies]
+grpcio = { version = "0.4", default-features = false, features = ["protobuf-codec"] }
+protobuf = "2.0.2"
+protos = { path = "../protos" }
+slot_clock = { path = "../beacon_chain/utils/slot_clock" }
+spec = { path = "../beacon_chain/spec" }
+tokio = "0.1.14"
+types = { path = "../beacon_chain/types" }
+slog = "^2.2.3"
+slog-term = "^2.4.0"
+slog-async = "^2.3.0"
+ssz = { path = "../beacon_chain/utils/ssz" }

validator_client/src/block_producer/mod.rs

@@ -0,0 +1,127 @@
+mod traits;
+
+use self::traits::{BeaconNode, BeaconNodeError};
+use crate::EpochDuties;
+use slot_clock::SlotClock;
+use spec::ChainSpec;
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+use types::BeaconBlock;
+
+#[derive(Debug, PartialEq)]
+pub enum PollOutcome {
+    BlockProduced,
+    SlashableBlockNotProduced,
+    BlockProductionNotRequired,
+    ProducerDutiesUnknown,
+    SlotAlreadyProcessed,
+    BeaconNodeUnableToProduceBlock,
+}
+
+#[derive(Debug, PartialEq)]
+pub enum PollError {
+    SlotClockError,
+    SlotUnknowable,
+    EpochMapPoisoned,
+    SlotClockPoisoned,
+    BeaconNodeError(BeaconNodeError),
+}
+
+pub struct BlockProducer<T: SlotClock, U: BeaconNode> {
+    pub last_processed_slot: u64,
+    _spec: Arc<ChainSpec>,
+    epoch_map: Arc<RwLock<HashMap<u64, EpochDuties>>>,
+    slot_clock: Arc<RwLock<T>>,
+    beacon_node: U,
+}
+
+impl<T: SlotClock, U: BeaconNode> BlockProducer<T, U> {
+    pub fn new(
+        spec: Arc<ChainSpec>,
+        epoch_map: Arc<RwLock<HashMap<u64, EpochDuties>>>,
+        slot_clock: Arc<RwLock<T>>,
+        beacon_node: U,
+    ) -> Self {
+        Self {
+            last_processed_slot: 0,
+            _spec: spec,
+            epoch_map,
+            slot_clock,
+            beacon_node,
+        }
+    }
+}
+
+impl<T: SlotClock, U: BeaconNode> BlockProducer<T, U> {
+    /// "Poll" to see if the validator is required to take any action.
+    ///
+    /// The slot clock will be read and any new actions undertaken.
+    pub fn poll(&mut self) -> Result<PollOutcome, PollError> {
+        let slot = self
+            .slot_clock
+            .read()
+            .map_err(|_| PollError::SlotClockPoisoned)?
+            .present_slot()
+            .map_err(|_| PollError::SlotClockError)?
+            .ok_or(PollError::SlotUnknowable)?;
+
+        // If this is a new slot.
+        if slot > self.last_processed_slot {
+            let is_block_production_slot = {
+                let epoch_map = self
+                    .epoch_map
+                    .read()
+                    .map_err(|_| PollError::EpochMapPoisoned)?;
+                match epoch_map.get(&slot) {
+                    None => return Ok(PollOutcome::ProducerDutiesUnknown),
+                    Some(duties) => duties.is_block_production_slot(slot)
+                }
+            };
+
+            if is_block_production_slot {
+                self.last_processed_slot = slot;
+
+                self.produce_block(slot)
+            } else {
+                Ok(PollOutcome::BlockProductionNotRequired)
+            }
+        } else {
+            Ok(PollOutcome::SlotAlreadyProcessed)
+        }
+    }
+
+    fn produce_block(&mut self, slot: u64) -> Result<PollOutcome, PollError> {
+        if let Some(block) = self.beacon_node.produce_beacon_block(slot)? {
+            if self.safe_to_produce(&block) {
+                let block = self.sign_block(block);
+                self.beacon_node.publish_beacon_block(block)?;
+                Ok(PollOutcome::BlockProduced)
+            } else {
+                Ok(PollOutcome::SlashableBlockNotProduced)
+            }
+        } else {
+            Ok(PollOutcome::BeaconNodeUnableToProduceBlock)
+        }
+    }
+
+    fn sign_block(&mut self, block: BeaconBlock) -> BeaconBlock {
+        // TODO: sign the block
+        self.store_produce(&block);
+        block
+    }
+
+    fn safe_to_produce(&self, _block: &BeaconBlock) -> bool {
+        // TODO: ensure the producer doesn't produce slashable blocks.
+        true
+    }
+
+    fn store_produce(&mut self, _block: &BeaconBlock) {
+        // TODO: record this block production to prevent future slashings.
+    }
+}
+
+impl From<BeaconNodeError> for PollError {
+    fn from(e: BeaconNodeError) -> PollError {
+        PollError::BeaconNodeError(e)
+    }
+}

validator_client/src/block_producer/traits.rs

@@ -0,0 +1,73 @@
+use protos::services::{
+    BeaconBlock as GrpcBeaconBlock, ProduceBeaconBlockRequest, PublishBeaconBlockRequest,
+};
+use protos::services_grpc::BeaconBlockServiceClient;
+use ssz::{ssz_encode, Decodable};
+use types::{BeaconBlock, BeaconBlockBody, Hash256, Signature};
+
+#[derive(Debug, PartialEq)]
+pub enum BeaconNodeError {
+    RemoteFailure(String),
+    DecodeFailure,
+}
+
+pub trait BeaconNode {
+    fn produce_beacon_block(&self, slot: u64) -> Result<Option<BeaconBlock>, BeaconNodeError>;
+    fn publish_beacon_block(&self, block: BeaconBlock) -> Result<bool, BeaconNodeError>;
+}
+
+impl BeaconNode for BeaconBlockServiceClient {
+    fn produce_beacon_block(&self, slot: u64) -> Result<Option<BeaconBlock>, BeaconNodeError> {
+        let mut req = ProduceBeaconBlockRequest::new();
+        req.set_slot(slot);
+
+        let reply = self
+            .produce_beacon_block(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        if reply.has_block() {
+            let block = reply.get_block();
+
+            let (signature, _) = Signature::ssz_decode(block.get_signature(), 0)
+                .map_err(|_| BeaconNodeError::DecodeFailure)?;
+
+            // TODO: this conversion is incomplete; fix it.
+            Ok(Some(BeaconBlock {
+                slot: block.get_slot(),
+                parent_root: Hash256::zero(),
+                state_root: Hash256::zero(),
+                randao_reveal: Hash256::from(block.get_randao_reveal()),
+                candidate_pow_receipt_root: Hash256::zero(),
+                signature,
+                body: BeaconBlockBody {
+                    proposer_slashings: vec![],
+                    casper_slashings: vec![],
+                    attestations: vec![],
+                    deposits: vec![],
+                    exits: vec![],
+                },
+            }))
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn publish_beacon_block(&self, block: BeaconBlock) -> Result<bool, BeaconNodeError> {
+        let mut req = PublishBeaconBlockRequest::new();
+
+        // TODO: this conversion is incomplete; fix it.
+        let mut grpc_block = GrpcBeaconBlock::new();
+        grpc_block.set_slot(block.slot);
+        grpc_block.set_block_root(vec![0]);
+        grpc_block.set_randao_reveal(block.randao_reveal.to_vec());
+        grpc_block.set_signature(ssz_encode(&block.signature));
+
+        req.set_block(grpc_block);
+
+        let reply = self
+            .publish_beacon_block(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        Ok(reply.get_success())
+    }
+}

validator_client/src/main.rs

@@ -0,0 +1,102 @@
+mod block_producer;
+
+use spec::ChainSpec;
+use tokio::prelude::*;
+use tokio::timer::Interval;
+
+use crate::block_producer::{BlockProducer, PollOutcome as BlockProducerPollOutcome};
+
+use std::time::{Duration, Instant};
+
+use std::sync::{Arc, RwLock};
+
+use std::collections::HashMap;
+
+use slot_clock::SystemTimeSlotClock;
+
+use grpcio::{ChannelBuilder, EnvBuilder};
+use protos::services_grpc::BeaconBlockServiceClient;
+
+use slog::{error, info, o, warn, Drain};
+
+fn main() {
+    // gRPC
+    let env = Arc::new(EnvBuilder::new().build());
+    let ch = ChannelBuilder::new(env).connect("localhost:50051");
+    let client = BeaconBlockServiceClient::new(ch);
+
+    // Logging
+    let decorator = slog_term::TermDecorator::new().build();
+    let drain = slog_term::CompactFormat::new(decorator).build().fuse();
+    let drain = slog_async::Async::new(drain).build().fuse();
+    let log = slog::Logger::root(drain, o!());
+
+    // Ethereum
+    let spec = Arc::new(ChainSpec::foundation());
+
+    let duration = spec
+        .slot_duration
+        .checked_mul(1_000)
+        .expect("Slot duration overflow when converting from seconds to millis.");
+
+    let epoch_map = Arc::new(RwLock::new(HashMap::new()));
+    let slot_clock = {
+        info!(log, "Genesis time"; "unix_epoch_seconds" => spec.genesis_time);
+        let clock = SystemTimeSlotClock::new(spec.genesis_time, spec.slot_duration)
+            .expect("Unable to instantiate SystemTimeSlotClock.");
+        Arc::new(RwLock::new(clock))
+    };
+
+    let mut block_producer =
+        BlockProducer::new(spec.clone(), epoch_map.clone(), slot_clock.clone(), client);
+
+    info!(log, "Slot duration"; "milliseconds" => duration);
+
+    let task = Interval::new(Instant::now(), Duration::from_millis(duration))
+        // .take(10)
+        .for_each(move |_instant| {
+            match block_producer.poll() {
+                Err(error) => {
+                    error!(log, "Block producer poll error"; "error" => format!("{:?}", error))
+                }
+                Ok(BlockProducerPollOutcome::BlockProduced) => info!(log, "Produced block"),
+                Ok(BlockProducerPollOutcome::SlashableBlockNotProduced) => {
+                    warn!(log, "Slashable block was not signed")
+                }
+                Ok(BlockProducerPollOutcome::BlockProductionNotRequired) => {
+                    info!(log, "Block production not required")
+                }
+                Ok(BlockProducerPollOutcome::ProducerDutiesUnknown) => {
+                    error!(log, "Block production duties unknown")
+                }
+                Ok(BlockProducerPollOutcome::SlotAlreadyProcessed) => {
+                    warn!(log, "Attempted to re-process slot")
+                }
+                Ok(BlockProducerPollOutcome::BeaconNodeUnableToProduceBlock) => {
+                    error!(log, "Beacon node unable to produce block")
+                }
+            };
+            Ok(())
+        })
+        .map_err(|e| panic!("Block producer interval errored; err={:?}", e));
+
+    tokio::run(task);
+}
+
+pub struct EpochDuties {
+    block_production_slot: Option<u64>,
+    shard: Option<u64>,
+}
+
+impl EpochDuties {
+    pub fn is_block_production_slot(&self, slot: u64) -> bool {
+        match self.block_production_slot {
+            Some(s) if s == slot => true,
+            _ => false,
+        }
+    }
+
+    pub fn has_shard(&self) -> bool {
+        self.shard.is_some()
+    }
+}