Fix compile errors from block_producer upgrades

eth2/block_producer/src/lib.rs

@@ -4,7 +4,7 @@ mod traits;
 use slot_clock::SlotClock;
 use spec::ChainSpec;
 use ssz::ssz_encode;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use types::{BeaconBlock, Hash256, ProposalSignedData, PublicKey};
 
 pub use self::traits::{
@@ -90,10 +90,6 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> BlockProducer<T, U
             .map_err(|_| Error::SlotClockError)?
             .ok_or(Error::SlotUnknowable)?;
 
-        let epoch = slot
-            .checked_div(self.spec.epoch_length)
-            .ok_or(Error::EpochLengthIsZero)?;
-
         // If this is a new slot.
         if !self.is_processed_slot(slot) {
             let is_block_production_slot = match self.epoch_map.is_block_production_slot(slot) {
@@ -104,6 +100,7 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> BlockProducer<T, U
                 Err(DutiesReaderError::UnknownValidator) => {
                     return Ok(PollOutcome::ValidatorIsUnknown(slot))
                 }
+                Err(DutiesReaderError::EpochLengthIsZero) => return Err(Error::EpochLengthIsZero),
                 Err(DutiesReaderError::Poisoned) => return Err(Error::EpochMapPoisoned),
             };
 
@@ -121,7 +118,7 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> BlockProducer<T, U
 
     fn is_processed_slot(&self, slot: u64) -> bool {
         match self.last_processed_slot {
-            Some(processed_slot) if processed_slot <= slot => true,
+            Some(processed_slot) if processed_slot >= slot => true,
             _ => false,
         }
     }
@@ -249,18 +246,20 @@ mod tests {
         let mut rng = XorShiftRng::from_seed([42; 16]);
 
         let spec = Arc::new(ChainSpec::foundation());
-        let slot_clock = Arc::new(RwLock::new(TestingSlotClock::new(0)));
+        let slot_clock = Arc::new(TestingSlotClock::new(0));
         let beacon_node = Arc::new(TestBeaconNode::default());
         let signer = Arc::new(TestSigner::new(Keypair::random()));
 
-        let mut epoch_map = TestEpochMap::new();
+        let mut epoch_map = TestEpochMap::new(spec.epoch_length);
         let produce_slot = 100;
         let produce_epoch = produce_slot / spec.epoch_length;
-        epoch_map.insert(produce_epoch, produce_slot);
+        epoch_map.map.insert(produce_epoch, produce_slot);
         let epoch_map = Arc::new(epoch_map);
+        let keypair = Keypair::random();
 
         let mut block_producer = BlockProducer::new(
             spec.clone(),
+            keypair.pk.clone(),
             epoch_map.clone(),
             slot_clock.clone(),
             beacon_node.clone(),
@@ -269,31 +268,32 @@ mod tests {
 
         // Configure responses from the BeaconNode.
         beacon_node.set_next_produce_result(Ok(Some(BeaconBlock::random_for_test(&mut rng))));
-        beacon_node.set_next_publish_result(Ok(true));
+        beacon_node.set_next_publish_result(Ok(PublishOutcome::ValidBlock));
+        beacon_node.set_next_nonce_result(Ok(0));
 
         // One slot before production slot...
-        slot_clock.write().unwrap().set_slot(produce_slot - 1);
+        slot_clock.set_slot(produce_slot - 1);
         assert_eq!(
             block_producer.poll(),
             Ok(PollOutcome::BlockProductionNotRequired(produce_slot - 1))
         );
 
         // On the produce slot...
-        slot_clock.write().unwrap().set_slot(produce_slot);
+        slot_clock.set_slot(produce_slot);
         assert_eq!(
             block_producer.poll(),
             Ok(PollOutcome::BlockProduced(produce_slot))
         );
 
         // Trying the same produce slot again...
-        slot_clock.write().unwrap().set_slot(produce_slot);
+        slot_clock.set_slot(produce_slot);
         assert_eq!(
             block_producer.poll(),
             Ok(PollOutcome::SlotAlreadyProcessed(produce_slot))
         );
 
         // One slot after the produce slot...
-        slot_clock.write().unwrap().set_slot(produce_slot + 1);
+        slot_clock.set_slot(produce_slot + 1);
         assert_eq!(
             block_producer.poll(),
             Ok(PollOutcome::BlockProductionNotRequired(produce_slot + 1))
@@ -301,7 +301,7 @@ mod tests {
 
         // In an epoch without known duties...
         let slot = (produce_epoch + 1) * spec.epoch_length;
-        slot_clock.write().unwrap().set_slot(slot);
+        slot_clock.set_slot(slot);
         assert_eq!(
             block_producer.poll(),
             Ok(PollOutcome::ProducerDutiesUnknown(slot))

eth2/block_producer/src/test_utils/epoch_map.rs

@@ -3,11 +3,11 @@ use std::collections::HashMap;
 
 pub struct TestEpochMap {
     epoch_length: u64,
-    map: HashMap<u64, u64>,
+    pub map: HashMap<u64, u64>,
 }
 
 impl TestEpochMap {
-    fn new(epoch_length: u64) -> Self {
+    pub fn new(epoch_length: u64) -> Self {
         Self {
             epoch_length,
             map: HashMap::new(),

eth2/block_producer/src/traits.rs

@@ -36,6 +36,7 @@ pub trait BeaconNode: Send + Sync {
 pub enum DutiesReaderError {
     UnknownValidator,
     UnknownEpoch,
+    EpochLengthIsZero,
     Poisoned,
 }
 

validator_client/src/block_producer_service/grpc.rs

@@ -1,11 +1,11 @@
-use block_producer::{BeaconNode, BeaconNodeError};
+use block_producer::{BeaconNode, BeaconNodeError, PublishOutcome};
 use protos::services::{
     BeaconBlock as GrpcBeaconBlock, ProduceBeaconBlockRequest, PublishBeaconBlockRequest,
 };
 use protos::services_grpc::BeaconBlockServiceClient;
 use ssz::{ssz_encode, Decodable};
 use std::sync::Arc;
-use types::{BeaconBlock, BeaconBlockBody, Eth1Data, Hash256, Signature};
+use types::{BeaconBlock, BeaconBlockBody, Eth1Data, Hash256, PublicKey, Signature};
 
 /// A newtype designed to wrap the gRPC-generated service so the `BeaconNode` trait may be
 /// implemented upon it.
@@ -20,11 +20,21 @@ impl BeaconBlockGrpcClient {
 }
 
 impl BeaconNode for BeaconBlockGrpcClient {
+    fn proposer_nonce(&self, pubkey: &PublicKey) -> Result<u64, BeaconNodeError> {
+        // TODO: this might not be required.
+        //
+        // See: https://github.com/ethereum/eth2.0-specs/pull/496
+        panic!("Not implemented.")
+    }
     /// Request a Beacon Node (BN) to produce a new block at the supplied slot.
     ///
     /// Returns `None` if it is not possible to produce at the supplied slot. For example, if the
     /// BN is unable to find a parent block.
-    fn produce_beacon_block(&self, slot: u64) -> Result<Option<BeaconBlock>, BeaconNodeError> {
+    fn produce_beacon_block(
+        &self,
+        slot: u64,
+        randao_reveal: &Signature,
+    ) -> Result<Option<BeaconBlock>, BeaconNodeError> {
         let mut req = ProduceBeaconBlockRequest::new();
         req.set_slot(slot);
 
@@ -73,7 +83,7 @@ impl BeaconNode for BeaconBlockGrpcClient {
     ///
     /// Generally, this will be called after a `produce_beacon_block` call with a block that has
     /// been completed (signed) by the validator client.
-    fn publish_beacon_block(&self, block: BeaconBlock) -> Result<bool, BeaconNodeError> {
+    fn publish_beacon_block(&self, block: BeaconBlock) -> Result<PublishOutcome, BeaconNodeError> {
         let mut req = PublishBeaconBlockRequest::new();
 
         // TODO: this conversion is incomplete; fix it.
@@ -90,6 +100,11 @@ impl BeaconNode for BeaconBlockGrpcClient {
             .publish_beacon_block(&req)
             .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
 
-        Ok(reply.get_success())
+        if reply.get_success() {
+            Ok(PublishOutcome::ValidBlock)
+        } else {
+            // TODO: distinguish between different errors
+            Ok(PublishOutcome::InvalidBlock("Publish failed".to_string()))
+        }
     }
 }

validator_client/src/block_producer_service/service.rs

@@ -42,6 +42,9 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> BlockProducerServi
                 Ok(BlockProducerPollOutcome::SignerRejection(slot)) => {
                     error!(self.log, "The cryptographic signer refused to sign the block"; "slot" => slot)
                 }
+                Ok(BlockProducerPollOutcome::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/duties/duties_map.rs

@@ -0,0 +1,81 @@
+use block_producer::{DutiesReader, DutiesReaderError};
+use std::collections::HashMap;
+use std::sync::RwLock;
+
+/// The information required for a validator to propose and attest during some epoch.
+///
+/// Generally obtained from a Beacon Node, this information contains the validators canonical index
+/// (thier sequence in the global validator induction process) and the "shuffling" for that index
+/// for some epoch.
+#[derive(Debug, PartialEq, Clone, Copy, Default)]
+pub struct EpochDuties {
+    pub validator_index: u64,
+    pub block_production_slot: Option<u64>,
+    // Future shard info
+}
+
+impl EpochDuties {
+    /// Returns `true` if the supplied `slot` is a slot in which the validator should produce a
+    /// block.
+    pub fn is_block_production_slot(&self, slot: u64) -> bool {
+        match self.block_production_slot {
+            Some(s) if s == slot => true,
+            _ => false,
+        }
+    }
+}
+
+pub enum EpochDutiesMapError {
+    Poisoned,
+}
+
+/// Maps an `epoch` to some `EpochDuties` for a single validator.
+pub struct EpochDutiesMap {
+    pub epoch_length: u64,
+    pub map: RwLock<HashMap<u64, EpochDuties>>,
+}
+
+impl EpochDutiesMap {
+    pub fn new(epoch_length: u64) -> Self {
+        Self {
+            epoch_length,
+            map: RwLock::new(HashMap::new()),
+        }
+    }
+
+    pub fn get(&self, epoch: u64) -> Result<Option<EpochDuties>, EpochDutiesMapError> {
+        let map = self.map.read().map_err(|_| EpochDutiesMapError::Poisoned)?;
+        match map.get(&epoch) {
+            Some(duties) => Ok(Some(duties.clone())),
+            None => Ok(None),
+        }
+    }
+
+    pub fn insert(
+        &self,
+        epoch: u64,
+        epoch_duties: EpochDuties,
+    ) -> Result<Option<EpochDuties>, EpochDutiesMapError> {
+        let mut map = self
+            .map
+            .write()
+            .map_err(|_| EpochDutiesMapError::Poisoned)?;
+        Ok(map.insert(epoch, epoch_duties))
+    }
+}
+
+impl DutiesReader for EpochDutiesMap {
+    fn is_block_production_slot(&self, slot: u64) -> Result<bool, DutiesReaderError> {
+        let epoch = slot
+            .checked_div(self.epoch_length)
+            .ok_or_else(|| DutiesReaderError::EpochLengthIsZero)?;
+
+        let map = self.map.read().map_err(|_| DutiesReaderError::Poisoned)?;
+        let duties = map
+            .get(&epoch)
+            .ok_or_else(|| DutiesReaderError::UnknownEpoch)?;
+        Ok(duties.is_block_production_slot(slot))
+    }
+}
+
+// TODO: add tests.

validator_client/src/duties/mod.rs

@@ -1,88 +1,18 @@
+mod duties_map;
 mod grpc;
 mod service;
 #[cfg(test)]
 mod test_node;
 mod traits;
 
+pub use self::duties_map::EpochDutiesMap;
+use self::duties_map::{EpochDuties, EpochDutiesMapError};
+pub use self::service::DutiesManagerService;
 use self::traits::{BeaconNode, BeaconNodeError};
-use block_producer::{DutiesReader, DutiesReaderError};
 use bls::PublicKey;
 use slot_clock::SlotClock;
 use spec::ChainSpec;
-use std::collections::HashMap;
+use std::sync::Arc;
-use std::sync::{Arc, RwLock};
-
-pub use self::service::DutiesManagerService;
-
-/// The information required for a validator to propose and attest during some epoch.
-///
-/// Generally obtained from a Beacon Node, this information contains the validators canonical index
-/// (thier sequence in the global validator induction process) and the "shuffling" for that index
-/// for some epoch.
-#[derive(Debug, PartialEq, Clone, Copy, Default)]
-pub struct EpochDuties {
-    pub validator_index: u64,
-    pub block_production_slot: Option<u64>,
-    // Future shard info
-}
-
-impl EpochDuties {
-    /// Returns `true` if the supplied `slot` is a slot in which the validator should produce a
-    /// block.
-    pub fn is_block_production_slot(&self, slot: u64) -> bool {
-        match self.block_production_slot {
-            Some(s) if s == slot => true,
-            _ => false,
-        }
-    }
-}
-
-pub enum EpochDutiesMapError {
-    Poisoned,
-}
-
-/// Maps an `epoch` to some `EpochDuties` for a single validator.
-pub struct EpochDutiesMap {
-    pub map: RwLock<HashMap<u64, EpochDuties>>,
-}
-
-impl EpochDutiesMap {
-    pub fn new() -> Self {
-        Self {
-            map: RwLock::new(HashMap::new()),
-        }
-    }
-
-    pub fn get(&self, epoch: u64) -> Result<Option<EpochDuties>, EpochDutiesMapError> {
-        let map = self.map.read().map_err(|_| EpochDutiesMapError::Poisoned)?;
-        match map.get(&epoch) {
-            Some(duties) => Ok(Some(duties.clone())),
-            None => Ok(None),
-        }
-    }
-
-    pub fn insert(
-        &self,
-        epoch: u64,
-        epoch_duties: EpochDuties,
-    ) -> Result<Option<EpochDuties>, EpochDutiesMapError> {
-        let mut map = self
-            .map
-            .write()
-            .map_err(|_| EpochDutiesMapError::Poisoned)?;
-        Ok(map.insert(epoch, epoch_duties))
-    }
-}
-
-impl DutiesReader for EpochDutiesMap {
-    fn is_block_production_slot(&self, epoch: u64, slot: u64) -> Result<bool, DutiesReaderError> {
-        let map = self.map.read().map_err(|_| DutiesReaderError::Poisoned)?;
-        let duties = map
-            .get(&epoch)
-            .ok_or_else(|| DutiesReaderError::UnknownEpoch)?;
-        Ok(duties.is_block_production_slot(slot))
-    }
-}
 
 #[derive(Debug, PartialEq, Clone, Copy)]
 pub enum PollOutcome {
@@ -117,7 +47,7 @@ pub struct DutiesManager<T: SlotClock, U: BeaconNode> {
     /// The validator's public key.
     pub pubkey: PublicKey,
     pub spec: Arc<ChainSpec>,
-    pub slot_clock: Arc<RwLock<T>>,
+    pub slot_clock: Arc<T>,
     pub beacon_node: Arc<U>,
 }
 
@@ -129,8 +59,6 @@ impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
     pub fn poll(&self) -> Result<PollOutcome, Error> {
         let slot = self
             .slot_clock
-            .read()
-            .map_err(|_| Error::SlotClockPoisoned)?
             .present_slot()
             .map_err(|_| Error::SlotClockError)?
             .ok_or(Error::SlotUnknowable)?;
@@ -187,9 +115,9 @@ mod tests {
     #[test]
     pub fn polling() {
         let spec = Arc::new(ChainSpec::foundation());
-        let duties_map = Arc::new(EpochDutiesMap::new());
+        let duties_map = Arc::new(EpochDutiesMap::new(spec.epoch_length));
         let keypair = Keypair::random();
-        let slot_clock = Arc::new(RwLock::new(TestingSlotClock::new(0)));
+        let slot_clock = Arc::new(TestingSlotClock::new(0));
         let beacon_node = Arc::new(TestBeaconNode::default());
 
         let manager = DutiesManager {

validator_client/src/main.rs

@@ -10,7 +10,7 @@ use slog::{error, info, o, Drain};
 use slot_clock::SystemTimeSlotClock;
 use spec::ChainSpec;
 use std::path::PathBuf;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::thread;
 
 mod block_producer_service;
@@ -92,7 +92,7 @@ fn main() {
         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))
+        Arc::new(clock)
     };
 
     let poll_interval_millis = spec.slot_duration * 1000 / 10; // 10% epoch time precision.
@@ -108,7 +108,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());
+        let duties_map = Arc::new(EpochDutiesMap::new(spec.epoch_length));
 
         // Spawn a new thread to maintain the validator's `EpochDuties`.
         let duties_manager_thread = {
@@ -139,6 +139,7 @@ fn main() {
         // Spawn a new thread to perform block production for the validator.
         let producer_thread = {
             let spec = spec.clone();
+            let pubkey = keypair.pk.clone();
             let signer = Arc::new(TestSigner::new(keypair.clone()));
             let duties_map = duties_map.clone();
             let slot_clock = slot_clock.clone();
@@ -146,7 +147,7 @@ fn main() {
             let client = Arc::new(BeaconBlockGrpcClient::new(beacon_block_grpc_client.clone()));
             thread::spawn(move || {
                 let block_producer =
-                    BlockProducer::new(spec, duties_map, slot_clock, client, signer);
+                    BlockProducer::new(spec, pubkey, duties_map, slot_clock, client, signer);
                 let mut block_producer_service = BlockProducerService {
                     block_producer,
                     poll_interval_millis,