Merge pull request #331 from sigp/age-validator-client

[WIP] Validator Client Update

beacon_node/Cargo.toml

@@ -14,6 +14,7 @@ slog-term = "^2.4.0"
 slog-async = "^2.3.0"
 ctrlc = { version = "3.1.1", features = ["termination"] }
 tokio = "0.1.15"
+tokio-timer = "0.2.10"
 futures = "0.1.25"
 exit-future = "0.1.3"
 state_processing = { path = "../eth2/state_processing" }

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -342,8 +342,17 @@ where
 
         // If required, transition the new state to the present slot.
         for _ in state.slot.as_u64()..present_slot.as_u64() {
+            // Ensure the next epoch state caches are built in case of an epoch transition.
+            state.build_epoch_cache(RelativeEpoch::NextWithoutRegistryChange, &self.spec)?;
+            state.build_epoch_cache(RelativeEpoch::NextWithRegistryChange, &self.spec)?;
+
             per_slot_processing(&mut *state, &latest_block_header, &self.spec)?;
         }
+        state.build_epoch_cache(RelativeEpoch::Previous, &self.spec)?;
+        state.build_epoch_cache(RelativeEpoch::Current, &self.spec)?;
+        state.build_epoch_cache(RelativeEpoch::NextWithoutRegistryChange, &self.spec)?;
+        state.build_epoch_cache(RelativeEpoch::NextWithRegistryChange, &self.spec)?;
+        state.update_pubkey_cache()?;
 
         Ok(())
     }
@@ -405,6 +414,20 @@ where
         }
     }
 
+    /// Reads the slot clock (see `self.read_slot_clock()` and returns the number of slots since
+    /// genesis.
+    pub fn slots_since_genesis(&self) -> Option<SlotHeight> {
+        let now = self.read_slot_clock()?;
+
+        if now < self.spec.genesis_slot {
+            None
+        } else {
+            Some(SlotHeight::from(
+                now.as_u64() - self.spec.genesis_slot.as_u64(),
+            ))
+        }
+    }
+
     /// Returns slot of the present state.
     ///
     /// This is distinct to `read_slot_clock`, which reads from the actual system clock. If
@@ -456,8 +479,8 @@ where
     }
 
     /// Produce an `AttestationData` that is valid for the present `slot` and given `shard`.
-    pub fn produce_attestation_data(&self, shard: u64) -> Result<AttestationData, Error> {
+    pub fn produce_attestation(&self, shard: u64) -> Result<AttestationData, Error> {
-        trace!("BeaconChain::produce_attestation_data: shard: {}", shard);
+        trace!("BeaconChain::produce_attestation: shard: {}", shard);
         let source_epoch = self.state.read().current_justified_epoch;
         let source_root = *self.state.read().get_block_root(
             source_epoch.start_slot(self.spec.slots_per_epoch),

beacon_node/beacon_chain/src/initialise.rs

@@ -28,15 +28,19 @@ pub fn initialise_beacon_chain(
     let block_store = Arc::new(BeaconBlockStore::new(db.clone()));
     let state_store = Arc::new(BeaconStateStore::new(db.clone()));
 
-    let state_builder = TestingBeaconStateBuilder::from_deterministic_keypairs(8, &spec);
+    let state_builder = TestingBeaconStateBuilder::from_default_keypairs_file_if_exists(8, &spec);
     let (genesis_state, _keypairs) = state_builder.build();
 
     let mut genesis_block = BeaconBlock::empty(&spec);
     genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
 
     // Slot clock
-    let slot_clock = SystemTimeSlotClock::new(genesis_state.genesis_time, spec.seconds_per_slot)
+    let slot_clock = SystemTimeSlotClock::new(
-        .expect("Unable to load SystemTimeSlotClock");
+        spec.genesis_slot,
+        genesis_state.genesis_time,
+        spec.seconds_per_slot,
+    )
+    .expect("Unable to load SystemTimeSlotClock");
     // Choose the fork choice
     let fork_choice = BitwiseLMDGhost::new(block_store.clone(), state_store.clone());
 
@@ -65,15 +69,19 @@ pub fn initialise_test_beacon_chain(
     let block_store = Arc::new(BeaconBlockStore::new(db.clone()));
     let state_store = Arc::new(BeaconStateStore::new(db.clone()));
 
-    let state_builder = TestingBeaconStateBuilder::from_deterministic_keypairs(8, spec);
+    let state_builder = TestingBeaconStateBuilder::from_default_keypairs_file_if_exists(8, spec);
     let (genesis_state, _keypairs) = state_builder.build();
 
     let mut genesis_block = BeaconBlock::empty(spec);
     genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
 
     // Slot clock
-    let slot_clock = SystemTimeSlotClock::new(genesis_state.genesis_time, spec.seconds_per_slot)
+    let slot_clock = SystemTimeSlotClock::new(
-        .expect("Unable to load SystemTimeSlotClock");
+        spec.genesis_slot,
+        genesis_state.genesis_time,
+        spec.seconds_per_slot,
+    )
+    .expect("Unable to load SystemTimeSlotClock");
     // Choose the fork choice
     let fork_choice = BitwiseLMDGhost::new(block_store.clone(), state_store.clone());
 

beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs

@@ -207,13 +207,13 @@ impl BeaconChainHarness {
     ///
     /// This is the ideal scenario for the Beacon Chain, 100% honest participation from
     /// validators.
-    pub fn advance_chain_with_block(&mut self) {
+    pub fn advance_chain_with_block(&mut self) -> BeaconBlock {
         self.increment_beacon_chain_slot();
 
         // Produce a new block.
         let block = self.produce_block();
         debug!("Submitting block for processing...");
-        match self.beacon_chain.process_block(block) {
+        match self.beacon_chain.process_block(block.clone()) {
             Ok(BlockProcessingOutcome::ValidBlock(_)) => {}
             other => panic!("block processing failed with {:?}", other),
         };
@@ -233,6 +233,8 @@ impl BeaconChainHarness {
         });
 
         debug!("Free attestations processed.");
+
+        block
     }
 
     /// Signs a message using some validators secret key with the `Fork` info from the latest state

beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs

@@ -50,18 +50,18 @@ impl<T: ClientDB, U: SlotClock, F: ForkChoice> DirectBeaconNode<T, U, F> {
 }
 
 impl<T: ClientDB, U: SlotClock, F: ForkChoice> AttesterBeaconNode for DirectBeaconNode<T, U, F> {
-    fn produce_attestation_data(
+    fn produce_attestation(
         &self,
         _slot: Slot,
         shard: u64,
     ) -> Result<Option<AttestationData>, NodeError> {
-        match self.beacon_chain.produce_attestation_data(shard) {
+        match self.beacon_chain.produce_attestation(shard) {
             Ok(attestation_data) => Ok(Some(attestation_data)),
             Err(e) => Err(NodeError::RemoteFailure(format!("{:?}", e))),
         }
     }
 
-    fn publish_attestation_data(
+    fn publish_attestation(
         &self,
         free_attestation: FreeAttestation,
     ) -> Result<AttestationPublishOutcome, NodeError> {

beacon_node/client/Cargo.toml

@@ -14,6 +14,7 @@ types = { path = "../../eth2/types" }
 slot_clock = { path = "../../eth2/utils/slot_clock" }
 error-chain = "0.12.0"
 slog = "^2.2.3"
+ssz = { path = "../../eth2/utils/ssz" }
 tokio = "0.1.15"
 clap = "2.32.0"
 dirs = "1.0.3"

beacon_node/client/src/lib.rs

@@ -8,12 +8,20 @@ pub mod notifier;
 use beacon_chain::BeaconChain;
 pub use client_config::ClientConfig;
 pub use client_types::ClientTypes;
+use db::ClientDB;
 use exit_future::Signal;
+use fork_choice::ForkChoice;
+use futures::{future::Future, Stream};
 use network::Service as NetworkService;
-use slog::o;
+use slog::{error, info, o};
+use slot_clock::SlotClock;
+use ssz::TreeHash;
 use std::marker::PhantomData;
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 use tokio::runtime::TaskExecutor;
+use tokio::timer::Interval;
+use types::Hash256;
 
 /// Main beacon node client service. This provides the connection and initialisation of the clients
 /// sub-services in multiple threads.
@@ -24,12 +32,10 @@ pub struct Client<T: ClientTypes> {
     beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
     /// Reference to the network service.
     pub network: Arc<NetworkService>,
-    /// Future to stop and begin shutdown of the Client.
+    /// Signal to terminate the RPC server.
-    //TODO: Decide best way to handle shutdown
+    pub rpc_exit_signal: Option<Signal>,
-    pub exit: exit_future::Exit,
+    /// Signal to terminate the slot timer.
-    /// The sending future to call to terminate the Client.
+    pub slot_timer_exit_signal: Option<Signal>,
-    //TODO: Decide best way to handle shutdown
-    pub exit_signal: Signal,
     /// The clients logger.
     log: slog::Logger,
     /// Marker to pin the beacon chain generics.
@@ -43,35 +49,128 @@ impl<TClientType: ClientTypes> Client<TClientType> {
         log: slog::Logger,
         executor: &TaskExecutor,
     ) -> error::Result<Self> {
-        let (exit_signal, exit) = exit_future::signal();
-
         // generate a beacon chain
         let beacon_chain = TClientType::initialise_beacon_chain(&config);
 
+        if beacon_chain.read_slot_clock().is_none() {
+            panic!("Cannot start client before genesis!")
+        }
+
+        // Block starting the client until we have caught the state up to the current slot.
+        //
+        // If we don't block here we create an initial scenario where we're unable to process any
+        // blocks and we're basically useless.
+        {
+            let state_slot = beacon_chain.state.read().slot;
+            let wall_clock_slot = beacon_chain.read_slot_clock().unwrap();
+            let slots_since_genesis = beacon_chain.slots_since_genesis().unwrap();
+            info!(
+                log,
+                "Initializing state";
+                "state_slot" => state_slot,
+                "wall_clock_slot" => wall_clock_slot,
+                "slots_since_genesis" => slots_since_genesis,
+                "catchup_distance" => wall_clock_slot - state_slot,
+            );
+        }
+        do_state_catchup(&beacon_chain, &log);
+        info!(
+            log,
+            "State initialized";
+            "state_slot" => beacon_chain.state.read().slot,
+            "wall_clock_slot" => beacon_chain.read_slot_clock().unwrap(),
+        );
+
         // Start the network service, libp2p and syncing threads
         // TODO: Add beacon_chain reference to network parameters
         let network_config = &config.net_conf;
         let network_logger = log.new(o!("Service" => "Network"));
-        let (network, _network_send) = NetworkService::new(
+        let (network, network_send) = NetworkService::new(
             beacon_chain.clone(),
             network_config,
             executor,
             network_logger,
         )?;
 
+        let mut rpc_exit_signal = None;
         // spawn the RPC server
         if config.rpc_conf.enabled {
-            rpc::start_server(&config.rpc_conf, &log);
+            rpc_exit_signal = Some(rpc::start_server(
+                &config.rpc_conf,
+                executor,
+                network_send,
+                beacon_chain.clone(),
+                &log,
+            ));
+        }
+
+        let (slot_timer_exit_signal, exit) = exit_future::signal();
+        if let Ok(Some(duration_to_next_slot)) = beacon_chain.slot_clock.duration_to_next_slot() {
+            // set up the validator work interval - start at next slot and proceed every slot
+            let interval = {
+                // Set the interval to start at the next slot, and every slot after
+                let slot_duration = Duration::from_secs(config.spec.seconds_per_slot);
+                //TODO: Handle checked add correctly
+                Interval::new(Instant::now() + duration_to_next_slot, slot_duration)
+            };
+
+            let chain = beacon_chain.clone();
+            let log = log.new(o!("Service" => "SlotTimer"));
+            executor.spawn(
+                exit.until(
+                    interval
+                        .for_each(move |_| {
+                            do_state_catchup(&chain, &log);
+
+                            Ok(())
+                        })
+                        .map_err(|_| ()),
+                )
+                .map(|_| ()),
+            );
         }
 
         Ok(Client {
             config,
             beacon_chain,
-            exit,
+            rpc_exit_signal,
-            exit_signal,
+            slot_timer_exit_signal: Some(slot_timer_exit_signal),
             log,
             network,
             phantom: PhantomData,
         })
     }
 }
+
+fn do_state_catchup<T, U, F>(chain: &Arc<BeaconChain<T, U, F>>, log: &slog::Logger)
+where
+    T: ClientDB,
+    U: SlotClock,
+    F: ForkChoice,
+{
+    if let Some(genesis_height) = chain.slots_since_genesis() {
+        let result = chain.catchup_state();
+
+        let common = o!(
+            "best_slot" => chain.head().beacon_block.slot,
+            "latest_block_root" => format!("{}", chain.head().beacon_block_root),
+            "wall_clock_slot" => chain.read_slot_clock().unwrap(),
+            "state_slot" => chain.state.read().slot,
+            "slots_since_genesis" => genesis_height,
+        );
+
+        match result {
+            Ok(_) => info!(
+                log,
+                "NewSlot";
+                common
+            ),
+            Err(e) => error!(
+                log,
+                "StateCatchupFailed";
+                "error" => format!("{:?}", e),
+                common
+            ),
+        };
+    }
+}

beacon_node/client/src/notifier.rs

@@ -2,7 +2,7 @@ use crate::Client;
 use crate::ClientTypes;
 use exit_future::Exit;
 use futures::{Future, Stream};
-use slog::{debug, info, o};
+use slog::{debug, o};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use tokio::runtime::TaskExecutor;
@@ -22,7 +22,7 @@ pub fn run<T: ClientTypes>(client: &Client<T>, executor: TaskExecutor, exit: Exi
 
     // build heartbeat logic here
     let heartbeat = move |_| {
-        info!(log, "Temp heartbeat output");
+        debug!(log, "Temp heartbeat output");
         //TODO: Remove this logic. Testing only
         let mut count = counter.lock().unwrap();
         *count += 1;

beacon_node/eth2-libp2p/src/behaviour.rs

@@ -13,10 +13,11 @@ use libp2p::{
     tokio_io::{AsyncRead, AsyncWrite},
     NetworkBehaviour, PeerId,
 };
-use slog::{debug, o};
+use slog::{debug, o, warn};
+use ssz::{ssz_encode, Decodable, DecodeError, Encodable, SszStream};
 use ssz_derive::{Decode, Encode};
 use types::Attestation;
-use types::Topic;
+use types::{Topic, TopicHash};
 
 /// Builds the network behaviour for the libp2p Swarm.
 /// Implements gossipsub message routing.
@@ -47,13 +48,36 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
 {
     fn inject_event(&mut self, event: GossipsubEvent) {
         match event {
-            GossipsubEvent::Message(message) => {
+            GossipsubEvent::Message(gs_msg) => {
-                let gs_message = String::from_utf8_lossy(&message.data);
+                debug!(self.log, "Received GossipEvent"; "msg" => format!("{:?}", gs_msg));
-                // TODO: Remove this type - debug only
+
-                self.events
+                let pubsub_message = match PubsubMessage::ssz_decode(&gs_msg.data, 0) {
-                    .push(BehaviourEvent::Message(gs_message.to_string()))
+                    //TODO: Punish peer on error
+                    Err(e) => {
+                        warn!(
+                            self.log,
+                            "Received undecodable message from Peer {:?} error", gs_msg.source;
+                            "error" => format!("{:?}", e)
+                        );
+                        return;
+                    }
+                    Ok((msg, _index)) => msg,
+                };
+
+                self.events.push(BehaviourEvent::GossipMessage {
+                    source: gs_msg.source,
+                    topics: gs_msg.topics,
+                    message: pubsub_message,
+                });
             }
-            _ => {}
+            GossipsubEvent::Subscribed {
+                peer_id: _,
+                topic: _,
+            }
+            | GossipsubEvent::Unsubscribed {
+                peer_id: _,
+                topic: _,
+            } => {}
         }
     }
 }
@@ -147,6 +171,14 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
     pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
         self.serenity_rpc.send_rpc(peer_id, rpc_event);
     }
+
+    /// Publishes a message on the pubsub (gossipsub) behaviour.
+    pub fn publish(&mut self, topics: Vec<Topic>, message: PubsubMessage) {
+        let message_bytes = ssz_encode(&message);
+        for topic in topics {
+            self.gossipsub.publish(topic, message_bytes.clone());
+        }
+    }
 }
 
 /// The types of events than can be obtained from polling the behaviour.
@@ -155,23 +187,73 @@ pub enum BehaviourEvent {
     PeerDialed(PeerId),
     Identified(PeerId, IdentifyInfo),
     // TODO: This is a stub at the moment
-    Message(String),
+    GossipMessage {
+        source: PeerId,
+        topics: Vec<TopicHash>,
+        message: PubsubMessage,
+    },
 }
 
-#[derive(Debug, Clone)]
+/// Messages that are passed to and from the pubsub (Gossipsub) behaviour.
-pub enum IncomingGossip {
+#[derive(Debug, Clone, PartialEq)]
-    Block(BlockGossip),
+pub enum PubsubMessage {
-    Attestation(AttestationGossip),
+    /// Gossipsub message providing notification of a new block.
+    Block(BlockRootSlot),
+    /// Gossipsub message providing notification of a new attestation.
+    Attestation(Attestation),
 }
 
-/// Gossipsub message providing notification of a new block.
+//TODO: Correctly encode/decode enums. Prefixing with integer for now.
-#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+impl Encodable for PubsubMessage {
-pub struct BlockGossip {
+    fn ssz_append(&self, s: &mut SszStream) {
-    pub root: BlockRootSlot,
+        match self {
+            PubsubMessage::Block(block_gossip) => {
+                0u32.ssz_append(s);
+                block_gossip.ssz_append(s);
+            }
+            PubsubMessage::Attestation(attestation_gossip) => {
+                1u32.ssz_append(s);
+                attestation_gossip.ssz_append(s);
+            }
+        }
+    }
 }
 
-/// Gossipsub message providing notification of a new attestation.
+impl Decodable for PubsubMessage {
-#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+    fn ssz_decode(bytes: &[u8], index: usize) -> Result<(Self, usize), DecodeError> {
-pub struct AttestationGossip {
+        let (id, index) = u32::ssz_decode(bytes, index)?;
-    pub attestation: Attestation,
+        match id {
+            0 => {
+                let (block, index) = BlockRootSlot::ssz_decode(bytes, index)?;
+                Ok((PubsubMessage::Block(block), index))
+            }
+            1 => {
+                let (attestation, index) = Attestation::ssz_decode(bytes, index)?;
+                Ok((PubsubMessage::Attestation(attestation), index))
+            }
+            _ => Err(DecodeError::Invalid),
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use types::*;
+
+    #[test]
+    fn ssz_encoding() {
+        let original = PubsubMessage::Block(BlockRootSlot {
+            block_root: Hash256::from_slice(&[42; 32]),
+            slot: Slot::new(4),
+        });
+
+        let encoded = ssz_encode(&original);
+
+        println!("{:?}", encoded);
+
+        let (decoded, _i) = PubsubMessage::ssz_decode(&encoded, 0).unwrap();
+
+        assert_eq!(original, decoded);
+    }
 }

beacon_node/eth2-libp2p/src/lib.rs

@@ -8,12 +8,13 @@ pub mod error;
 pub mod rpc;
 mod service;
 
+pub use behaviour::PubsubMessage;
 pub use config::Config as NetworkConfig;
 pub use libp2p::{
     gossipsub::{GossipsubConfig, GossipsubConfigBuilder},
     PeerId,
 };
-pub use rpc::{HelloMessage, RPCEvent};
+pub use rpc::RPCEvent;
 pub use service::Libp2pEvent;
 pub use service::Service;
 pub use types::multiaddr;

beacon_node/eth2-libp2p/src/service.rs

@@ -1,4 +1,4 @@
-use crate::behaviour::{Behaviour, BehaviourEvent};
+use crate::behaviour::{Behaviour, BehaviourEvent, PubsubMessage};
 use crate::error;
 use crate::multiaddr::Protocol;
 use crate::rpc::RPCEvent;
@@ -17,7 +17,7 @@ use libp2p::{core, secio, PeerId, Swarm, Transport};
 use slog::{debug, info, trace, warn};
 use std::io::{Error, ErrorKind};
 use std::time::Duration;
-use types::TopicBuilder;
+use types::{TopicBuilder, TopicHash};
 
 /// The configuration and state of the libp2p components for the beacon node.
 pub struct Service {
@@ -108,9 +108,17 @@ impl Stream for Service {
                 //Behaviour events
                 Ok(Async::Ready(Some(event))) => match event {
                     // TODO: Stub here for debugging
-                    BehaviourEvent::Message(m) => {
+                    BehaviourEvent::GossipMessage {
-                        debug!(self.log, "Message received: {}", m);
+                        source,
-                        return Ok(Async::Ready(Some(Libp2pEvent::Message(m))));
+                        topics,
+                        message,
+                    } => {
+                        debug!(self.log, "Pubsub message received: {:?}", message);
+                        return Ok(Async::Ready(Some(Libp2pEvent::PubsubMessage {
+                            source,
+                            topics,
+                            message,
+                        })));
                     }
                     BehaviourEvent::RPC(peer_id, event) => {
                         return Ok(Async::Ready(Some(Libp2pEvent::RPC(peer_id, event))));
@@ -172,6 +180,10 @@ pub enum Libp2pEvent {
     PeerDialed(PeerId),
     /// Received information about a peer on the network.
     Identified(PeerId, IdentifyInfo),
-    // TODO: Pub-sub testing only.
+    /// Received pubsub message.
-    Message(String),
+    PubsubMessage {
+        source: PeerId,
+        topics: Vec<TopicHash>,
+        message: PubsubMessage,
+    },
 }

beacon_node/network/src/beacon_chain.rs

@@ -7,7 +7,7 @@ use beacon_chain::{
     types::{BeaconState, ChainSpec},
     AggregationOutcome, CheckPoint,
 };
-use eth2_libp2p::HelloMessage;
+use eth2_libp2p::rpc::HelloMessage;
 use types::{Attestation, BeaconBlock, BeaconBlockBody, BeaconBlockHeader, Epoch, Hash256, Slot};
 
 pub use beacon_chain::{BeaconChainError, BlockProcessingOutcome};
@@ -109,7 +109,7 @@ where
         let state = self.get_state();
 
         HelloMessage {
-            network_id: spec.network_id,
+            network_id: spec.chain_id,
             latest_finalized_root: state.finalized_root,
             latest_finalized_epoch: state.finalized_epoch,
             best_root: self.best_block_root(),

beacon_node/network/src/lib.rs

@@ -6,4 +6,5 @@ pub mod service;
 pub mod sync;
 
 pub use eth2_libp2p::NetworkConfig;
+pub use service::NetworkMessage;
 pub use service::Service;

beacon_node/network/src/message_handler.rs

@@ -4,7 +4,7 @@ use crate::service::{NetworkMessage, OutgoingMessage};
 use crate::sync::SimpleSync;
 use crossbeam_channel::{unbounded as channel, Sender};
 use eth2_libp2p::{
-    behaviour::IncomingGossip,
+    behaviour::PubsubMessage,
     rpc::{methods::GoodbyeReason, RPCRequest, RPCResponse, RequestId},
     PeerId, RPCEvent,
 };
@@ -41,7 +41,7 @@ pub enum HandlerMessage {
     /// An RPC response/request has been received.
     RPC(PeerId, RPCEvent),
     /// A gossip message has been received.
-    IncomingGossip(PeerId, IncomingGossip),
+    PubsubMessage(PeerId, PubsubMessage),
 }
 
 impl MessageHandler {
@@ -92,7 +92,7 @@ impl MessageHandler {
                 self.handle_rpc_message(peer_id, rpc_event);
             }
             // we have received an RPC message request/response
-            HandlerMessage::IncomingGossip(peer_id, gossip) => {
+            HandlerMessage::PubsubMessage(peer_id, gossip) => {
                 self.handle_gossip(peer_id, gossip);
             }
             //TODO: Handle all messages
@@ -205,13 +205,13 @@ impl MessageHandler {
     }
 
     /// Handle RPC messages
-    fn handle_gossip(&mut self, peer_id: PeerId, gossip_message: IncomingGossip) {
+    fn handle_gossip(&mut self, peer_id: PeerId, gossip_message: PubsubMessage) {
         match gossip_message {
-            IncomingGossip::Block(message) => {
+            PubsubMessage::Block(message) => {
                 self.sync
                     .on_block_gossip(peer_id, message, &mut self.network_context)
             }
-            IncomingGossip::Attestation(message) => {
+            PubsubMessage::Attestation(message) => {
                 self.sync
                     .on_attestation_gossip(peer_id, message, &mut self.network_context)
             }

beacon_node/network/src/service.rs

@@ -3,15 +3,16 @@ use crate::error;
 use crate::message_handler::{HandlerMessage, MessageHandler};
 use crate::NetworkConfig;
 use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
-use eth2_libp2p::RPCEvent;
 use eth2_libp2p::Service as LibP2PService;
 use eth2_libp2p::{Libp2pEvent, PeerId};
+use eth2_libp2p::{PubsubMessage, RPCEvent};
 use futures::prelude::*;
 use futures::sync::oneshot;
 use futures::Stream;
 use slog::{debug, info, o, trace};
 use std::sync::Arc;
 use tokio::runtime::TaskExecutor;
+use types::Topic;
 
 /// Service that handles communication between internal services and the eth2_libp2p network service.
 pub struct Service {
@@ -99,6 +100,7 @@ fn spawn_service(
     Ok(network_exit)
 }
 
+//TODO: Potentially handle channel errors
 fn network_service(
     mut libp2p_service: LibP2PService,
     network_recv: crossbeam_channel::Receiver<NetworkMessage>,
@@ -128,10 +130,17 @@ fn network_service(
                             "We have identified peer: {:?} with {:?}", peer_id, info
                         );
                     }
-                    Libp2pEvent::Message(m) => debug!(
+                    Libp2pEvent::PubsubMessage {
-                        libp2p_service.log,
+                        source,
-                        "Network Service: Message received: {}", m
+                        topics: _,
-                    ),
+                        message,
+                    } => {
+                        //TODO: Decide if we need to propagate the topic upwards. (Potentially for
+                        //attestations)
+                        message_handler_send
+                            .send(HandlerMessage::PubsubMessage(source, message))
+                            .map_err(|_| " failed to send pubsub message to handler")?;
+                    }
                 },
                 Ok(Async::Ready(None)) => unreachable!("Stream never ends"),
                 Ok(Async::NotReady) => break,
@@ -156,6 +165,10 @@ fn network_service(
                         }
                     };
                 }
+                Ok(NetworkMessage::Publish { topics, message }) => {
+                    debug!(log, "Sending pubsub message on topics {:?}", topics);
+                    libp2p_service.swarm.publish(topics, message);
+                }
                 Err(TryRecvError::Empty) => break,
                 Err(TryRecvError::Disconnected) => {
                     return Err(eth2_libp2p::error::Error::from(
@@ -174,6 +187,11 @@ pub enum NetworkMessage {
     /// Send a message to libp2p service.
     //TODO: Define typing for messages across the wire
     Send(PeerId, OutgoingMessage),
+    /// Publish a message to pubsub mechanism.
+    Publish {
+        topics: Vec<Topic>,
+        message: PubsubMessage,
+    },
 }
 
 /// Type of outgoing messages that can be sent through the network service.

beacon_node/network/src/sync/simple_sync.rs

@@ -1,7 +1,6 @@
 use super::import_queue::ImportQueue;
 use crate::beacon_chain::BeaconChain;
 use crate::message_handler::NetworkContext;
-use eth2_libp2p::behaviour::{AttestationGossip, BlockGossip};
 use eth2_libp2p::rpc::methods::*;
 use eth2_libp2p::rpc::{RPCRequest, RPCResponse, RequestId};
 use eth2_libp2p::PeerId;
@@ -9,7 +8,7 @@ use slog::{debug, error, info, o, warn};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
-use types::{Epoch, Hash256, Slot};
+use types::{Attestation, Epoch, Hash256, Slot};
 
 /// The number of slots that we can import blocks ahead of us, before going into full Sync mode.
 const SLOT_IMPORT_TOLERANCE: u64 = 100;
@@ -521,12 +520,12 @@ impl SimpleSync {
     pub fn on_block_gossip(
         &mut self,
         peer_id: PeerId,
-        msg: BlockGossip,
+        msg: BlockRootSlot,
         network: &mut NetworkContext,
     ) {
-        debug!(
+        info!(
             self.log,
-            "BlockGossip";
+            "NewGossipBlock";
             "peer" => format!("{:?}", peer_id),
         );
         // TODO: filter out messages that a prior to the finalized slot.
@@ -535,12 +534,12 @@ impl SimpleSync {
         // now.
         //
         // Note: only requests the new block -- will fail if we don't have its parents.
-        if self.import_queue.is_new_block(&msg.root.block_root) {
+        if self.import_queue.is_new_block(&msg.block_root) {
             self.request_block_headers(
                 peer_id,
                 BeaconBlockHeadersRequest {
-                    start_root: msg.root.block_root,
+                    start_root: msg.block_root,
-                    start_slot: msg.root.slot,
+                    start_slot: msg.slot,
                     max_headers: 1,
                     skip_slots: 0,
                 },
@@ -555,19 +554,19 @@ impl SimpleSync {
     pub fn on_attestation_gossip(
         &mut self,
         peer_id: PeerId,
-        msg: AttestationGossip,
+        msg: Attestation,
         _network: &mut NetworkContext,
     ) {
-        debug!(
+        info!(
             self.log,
-            "AttestationGossip";
+            "NewAttestationGossip";
             "peer" => format!("{:?}", peer_id),
         );
 
         // Awaiting a proper operations pool before we can import attestations.
         //
         // https://github.com/sigp/lighthouse/issues/281
-        match self.chain.process_attestation(msg.attestation) {
+        match self.chain.process_attestation(msg) {
             Ok(_) => panic!("Impossible, method not implemented."),
             Err(_) => error!(self.log, "Attestation processing not implemented!"),
         }

beacon_node/rpc/Cargo.toml

@@ -7,7 +7,12 @@ edition = "2018"
 [dependencies]
 bls = { path = "../../eth2/utils/bls" }
 beacon_chain = { path = "../beacon_chain" }
-
+network = { path = "../network" }
+eth2-libp2p = { path = "../eth2-libp2p" }
+version = { path = "../version" }
+types = { path = "../../eth2/types" }
+ssz = { path = "../../eth2/utils/ssz" }
+slot_clock = { path = "../../eth2/utils/slot_clock" }
 protos = { path = "../../protos" }
 grpcio = { version = "0.4", default-features = false, features = ["protobuf-codec"] }
 protobuf = "2.0.2"
@@ -16,8 +21,8 @@ db = { path = "../db" }
 dirs = "1.0.3"
 futures = "0.1.23"
 slog = "^2.2.3"
-slot_clock = { path = "../../eth2/utils/slot_clock" }
 slog-term = "^2.4.0"
 slog-async = "^2.3.0"
-types = { path = "../../eth2/types" }
+tokio = "0.1.17"
-ssz = { path = "../../eth2/utils/ssz" }
+exit-future = "0.1.4"
+crossbeam-channel = "0.3.8"

beacon_node/rpc/src/beacon_attester.rs

@@ -0,0 +1,61 @@
+use futures::Future;
+use grpcio::{RpcContext, UnarySink};
+use protos::services::{
+    Attestation as AttestationProto, ProduceAttestation, ProduceAttestationResponse,
+    ProduceAttestationRequest, PublishAttestationResponse, PublishAttestationRequest,
+    PublishAttestation
+};
+use protos::services_grpc::BeaconBlockService;
+use slog::Logger;
+
+#[derive(Clone)]
+pub struct AttestationServiceInstance {
+    pub log: Logger,
+}
+
+impl AttestationService for AttestationServiceInstance {
+    /// Produce a `BeaconBlock` for signing by a validator.
+    fn produce_attestation(
+        &mut self,
+        ctx: RpcContext,
+        req: ProduceAttestationRequest,
+        sink: UnarySink<ProduceAttestationResponse>,
+    ) {
+        println!("producing attestation at slot {}", req.get_slot());
+
+        // TODO: build a legit block.
+        let mut attestation = AttestationProto::new();
+        attestation.set_slot(req.get_slot());
+        // TODO Set the shard to something legit.
+        attestation.set_shard(0);
+        attestation.set_block_root(b"cats".to_vec());
+
+        let mut resp = ProduceAttestationResponse::new();
+        resp.set_attestation_data(attestation);
+
+        let f = sink
+            .success(resp)
+            .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
+        ctx.spawn(f)
+    }
+
+    /// Accept some fully-formed `BeaconBlock`, process and publish it.
+    fn publish_attestation(
+        &mut self,
+        ctx: RpcContext,
+        req: PublishAttestationRequest,
+        sink: UnarySink<PublishAttestationResponse>,
+    ) {
+        println!("publishing attestation {:?}", req.get_block());
+
+        // TODO: actually process the block.
+        let mut resp = PublishAttestationResponse::new();
+
+        resp.set_success(true);
+
+        let f = sink
+            .success(resp)
+            .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
+        ctx.spawn(f)
+    }
+}

beacon_node/rpc/src/beacon_block.rs

@@ -1,14 +1,25 @@
+use crate::beacon_chain::BeaconChain;
+use crossbeam_channel;
+use eth2_libp2p::rpc::methods::BlockRootSlot;
+use eth2_libp2p::PubsubMessage;
 use futures::Future;
 use grpcio::{RpcContext, UnarySink};
+use network::NetworkMessage;
 use protos::services::{
     BeaconBlock as BeaconBlockProto, ProduceBeaconBlockRequest, ProduceBeaconBlockResponse,
     PublishBeaconBlockRequest, PublishBeaconBlockResponse,
 };
 use protos::services_grpc::BeaconBlockService;
 use slog::Logger;
+use slog::{debug, error, info, warn};
+use ssz::{Decodable, TreeHash};
+use std::sync::Arc;
+use types::{BeaconBlock, Hash256, Slot};
 
 #[derive(Clone)]
 pub struct BeaconBlockServiceInstance {
+    pub chain: Arc<BeaconChain>,
+    pub network_chan: crossbeam_channel::Sender<NetworkMessage>,
     pub log: Logger,
 }
 
@@ -24,8 +35,7 @@ impl BeaconBlockService for BeaconBlockServiceInstance {
 
         // TODO: build a legit block.
         let mut block = BeaconBlockProto::new();
-        block.set_slot(req.get_slot());
+        block.set_ssz(b"cats".to_vec());
-        block.set_block_root(b"cats".to_vec());
 
         let mut resp = ProduceBeaconBlockResponse::new();
         resp.set_block(block);
@@ -43,11 +53,88 @@ impl BeaconBlockService for BeaconBlockServiceInstance {
         req: PublishBeaconBlockRequest,
         sink: UnarySink<PublishBeaconBlockResponse>,
     ) {
-        println!("publishing {:?}", req.get_block());
-
-        // TODO: actually process the block.
         let mut resp = PublishBeaconBlockResponse::new();
-        resp.set_success(true);
+
+        let ssz_serialized_block = req.get_block().get_ssz();
+
+        match BeaconBlock::ssz_decode(ssz_serialized_block, 0) {
+            Ok((block, _i)) => {
+                let block_root = Hash256::from_slice(&block.hash_tree_root()[..]);
+
+                match self.chain.process_block(block.clone()) {
+                    Ok(outcome) => {
+                        if outcome.sucessfully_processed() {
+                            // Block was successfully processed.
+                            info!(
+                                self.log,
+                                "PublishBeaconBlock";
+                                "type" => "valid_block",
+                                "block_slot" => block.slot,
+                                "outcome" => format!("{:?}", outcome)
+                            );
+
+                            // TODO: Obtain topics from the network service properly.
+                            let topic =
+                                types::TopicBuilder::new("beacon_chain".to_string()).build();
+                            let message = PubsubMessage::Block(BlockRootSlot {
+                                block_root,
+                                slot: block.slot,
+                            });
+
+                            println!("Sending beacon block to gossipsub");
+                            self.network_chan.send(NetworkMessage::Publish {
+                                topics: vec![topic],
+                                message,
+                            });
+
+                            resp.set_success(true);
+                        } else if outcome.is_invalid() {
+                            // Block was invalid.
+                            warn!(
+                                self.log,
+                                "PublishBeaconBlock";
+                                "type" => "invalid_block",
+                                "outcome" => format!("{:?}", outcome)
+                            );
+
+                            resp.set_success(false);
+                            resp.set_msg(
+                                format!("InvalidBlock: {:?}", outcome).as_bytes().to_vec(),
+                            );
+                        } else {
+                            // Some failure during processing.
+                            warn!(
+                                self.log,
+                                "PublishBeaconBlock";
+                                "type" => "unable_to_import",
+                                "outcome" => format!("{:?}", outcome)
+                            );
+
+                            resp.set_success(false);
+                            resp.set_msg(format!("other: {:?}", outcome).as_bytes().to_vec());
+                        }
+                    }
+                    Err(e) => {
+                        // Some failure during processing.
+                        error!(
+                            self.log,
+                            "PublishBeaconBlock";
+                            "type" => "failed_to_process",
+                            "error" => format!("{:?}", e)
+                        );
+
+                        resp.set_success(false);
+                        resp.set_msg(format!("failed_to_process: {:?}", e).as_bytes().to_vec());
+                    }
+                }
+
+                resp.set_success(true);
+            }
+            Err(_) => {
+                resp.set_success(false);
+                resp.set_msg(b"Invalid SSZ".to_vec());
+            }
+        };
 
         let f = sink
             .success(resp)

beacon_node/rpc/src/beacon_chain.rs

@@ -0,0 +1,42 @@
+use beacon_chain::BeaconChain as RawBeaconChain;
+use beacon_chain::{
+    db::ClientDB,
+    fork_choice::ForkChoice,
+    parking_lot::RwLockReadGuard,
+    slot_clock::SlotClock,
+    types::{BeaconState, ChainSpec},
+};
+pub use beacon_chain::{BeaconChainError, BlockProcessingOutcome};
+use types::BeaconBlock;
+
+/// The RPC's API to the beacon chain.
+pub trait BeaconChain: Send + Sync {
+    fn get_spec(&self) -> &ChainSpec;
+
+    fn get_state(&self) -> RwLockReadGuard<BeaconState>;
+
+    fn process_block(&self, block: BeaconBlock)
+        -> Result<BlockProcessingOutcome, BeaconChainError>;
+}
+
+impl<T, U, F> BeaconChain for RawBeaconChain<T, U, F>
+where
+    T: ClientDB + Sized,
+    U: SlotClock,
+    F: ForkChoice,
+{
+    fn get_spec(&self) -> &ChainSpec {
+        &self.spec
+    }
+
+    fn get_state(&self) -> RwLockReadGuard<BeaconState> {
+        self.state.read()
+    }
+
+    fn process_block(
+        &self,
+        block: BeaconBlock,
+    ) -> Result<BlockProcessingOutcome, BeaconChainError> {
+        self.process_block(block)
+    }
+}

beacon_node/rpc/src/beacon_node.rs

@@ -0,0 +1,47 @@
+use crate::beacon_chain::BeaconChain;
+use futures::Future;
+use grpcio::{RpcContext, UnarySink};
+use protos::services::{Empty, Fork, NodeInfoResponse};
+use protos::services_grpc::BeaconNodeService;
+use slog::{trace, warn};
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct BeaconNodeServiceInstance {
+    pub chain: Arc<BeaconChain>,
+    pub log: slog::Logger,
+}
+
+impl BeaconNodeService for BeaconNodeServiceInstance {
+    /// Provides basic node information.
+    fn info(&mut self, ctx: RpcContext, _req: Empty, sink: UnarySink<NodeInfoResponse>) {
+        trace!(self.log, "Node info requested via RPC");
+
+        // build the response
+        let mut node_info = NodeInfoResponse::new();
+        node_info.set_version(version::version());
+
+        // get the chain state
+        let state = self.chain.get_state();
+        let state_fork = state.fork.clone();
+        let genesis_time = state.genesis_time.clone();
+
+        // build the rpc fork struct
+        let mut fork = Fork::new();
+        fork.set_previous_version(state_fork.previous_version.to_vec());
+        fork.set_current_version(state_fork.current_version.to_vec());
+        fork.set_epoch(state_fork.epoch.into());
+
+        node_info.set_fork(fork);
+        node_info.set_genesis_time(genesis_time);
+        node_info.set_genesis_slot(self.chain.get_spec().genesis_slot.as_u64());
+        node_info.set_chain_id(self.chain.get_spec().chain_id as u32);
+
+        // send the node_info the requester
+        let error_log = self.log.clone();
+        let f = sink
+            .success(node_info)
+            .map_err(move |e| warn!(error_log, "failed to reply {:?}", e));
+        ctx.spawn(f)
+    }
+}

beacon_node/rpc/src/lib.rs

@@ -1,38 +1,85 @@
 mod beacon_block;
+pub mod beacon_chain;
+mod beacon_node;
 pub mod config;
 mod validator;
 
 use self::beacon_block::BeaconBlockServiceInstance;
+use self::beacon_chain::BeaconChain;
+use self::beacon_node::BeaconNodeServiceInstance;
 use self::validator::ValidatorServiceInstance;
 pub use config::Config as RPCConfig;
+use futures::{future, Future};
 use grpcio::{Environment, Server, ServerBuilder};
-use protos::services_grpc::{create_beacon_block_service, create_validator_service};
+use network::NetworkMessage;
+use protos::services_grpc::{
+    create_beacon_block_service, create_beacon_node_service, create_validator_service,
+};
+use slog::{info, o, warn};
 use std::sync::Arc;
+use tokio::runtime::TaskExecutor;
 
-use slog::{info, o};
+pub fn start_server(
-
+    config: &RPCConfig,
-pub fn start_server(config: &RPCConfig, log: &slog::Logger) -> Server {
+    executor: &TaskExecutor,
+    network_chan: crossbeam_channel::Sender<NetworkMessage>,
+    beacon_chain: Arc<BeaconChain>,
+    log: &slog::Logger,
+) -> exit_future::Signal {
     let log = log.new(o!("Service"=>"RPC"));
     let env = Arc::new(Environment::new(1));
 
+    // build a channel to kill the rpc server
+    let (rpc_exit_signal, rpc_exit) = exit_future::signal();
+
+    // build the individual rpc services
+    let beacon_node_service = {
+        let instance = BeaconNodeServiceInstance {
+            chain: beacon_chain.clone(),
+            log: log.clone(),
+        };
+        create_beacon_node_service(instance)
+    };
+
     let beacon_block_service = {
-        let instance = BeaconBlockServiceInstance { log: log.clone() };
+        let instance = BeaconBlockServiceInstance {
+            chain: beacon_chain.clone(),
+            network_chan,
+            log: log.clone(),
+        };
         create_beacon_block_service(instance)
     };
     let validator_service = {
-        let instance = ValidatorServiceInstance { log: log.clone() };
+        let instance = ValidatorServiceInstance {
+            chain: beacon_chain.clone(),
+            log: log.clone(),
+        };
         create_validator_service(instance)
     };
 
     let mut server = ServerBuilder::new(env)
         .register_service(beacon_block_service)
         .register_service(validator_service)
+        .register_service(beacon_node_service)
         .bind(config.listen_address.to_string(), config.port)
         .build()
         .unwrap();
-    server.start();
+
-    for &(ref host, port) in server.bind_addrs() {
+    let spawn_rpc = {
-        info!(log, "gRPC listening on {}:{}", host, port);
+        server.start();
-    }
+        for &(ref host, port) in server.bind_addrs() {
-    server
+            info!(log, "gRPC listening on {}:{}", host, port);
+        }
+        rpc_exit.and_then(move |_| {
+            info!(log, "RPC Server shutting down");
+            server
+                .shutdown()
+                .wait()
+                .map(|_| ())
+                .map_err(|e| warn!(log, "RPC server failed to shutdown: {:?}", e))?;
+            Ok(())
+        })
+    };
+    executor.spawn(spawn_rpc);
+    rpc_exit_signal
 }

beacon_node/rpc/src/validator.rs

@@ -1,60 +1,166 @@
+use crate::beacon_chain::BeaconChain;
 use bls::PublicKey;
 use futures::Future;
 use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink};
-use protos::services::{
+use protos::services::{ActiveValidator, GetDutiesRequest, GetDutiesResponse, ValidatorDuty};
-    IndexResponse, ProposeBlockSlotRequest, ProposeBlockSlotResponse, PublicKey as PublicKeyRequest,
-};
 use protos::services_grpc::ValidatorService;
-use slog::{debug, Logger};
+use slog::{debug, info, warn, Logger};
 use ssz::Decodable;
+use std::sync::Arc;
+use types::{Epoch, RelativeEpoch};
 
 #[derive(Clone)]
 pub struct ValidatorServiceInstance {
+    pub chain: Arc<BeaconChain>,
     pub log: Logger,
 }
+//TODO: Refactor Errors
 
 impl ValidatorService for ValidatorServiceInstance {
-    fn validator_index(
+    /// For a list of validator public keys, this function returns the slot at which each
+    /// validator must propose a block, attest to a shard, their shard committee and the shard they
+    /// need to attest to.
+    fn get_validator_duties(
         &mut self,
         ctx: RpcContext,
-        req: PublicKeyRequest,
+        req: GetDutiesRequest,
-        sink: UnarySink<IndexResponse>,
+        sink: UnarySink<GetDutiesResponse>,
     ) {
-        if let Ok((public_key, _)) = PublicKey::ssz_decode(req.get_public_key(), 0) {
+        let validators = req.get_validators();
-            debug!(self.log, "RPC request"; "endpoint" => "ValidatorIndex", "public_key" => public_key.concatenated_hex_id());
+        debug!(self.log, "RPC request"; "endpoint" => "GetValidatorDuties", "epoch" => req.get_epoch());
 
-            let mut resp = IndexResponse::new();
+        let epoch = Epoch::from(req.get_epoch());
+        let mut resp = GetDutiesResponse::new();
+        let resp_validators = resp.mut_active_validators();
 
-            // TODO: return a legit value.
+        let spec = self.chain.get_spec();
-            resp.set_index(1);
+        let state = self.chain.get_state();
 
-            let f = sink
+        let relative_epoch =
-                .success(resp)
+            match RelativeEpoch::from_epoch(state.slot.epoch(spec.slots_per_epoch), epoch) {
-                .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
+                Ok(v) => v,
-            ctx.spawn(f)
+                Err(e) => {
-        } else {
+                    // incorrect epoch
-            let f = sink
+                    let log_clone = self.log.clone();
-                .fail(RpcStatus::new(
+                    let f = sink
-                    RpcStatusCode::InvalidArgument,
+                        .fail(RpcStatus::new(
-                    Some("Invalid public_key".to_string()),
+                            RpcStatusCode::FailedPrecondition,
-                ))
+                            Some(format!("Invalid epoch: {:?}", e)),
-                .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
+                        ))
-            ctx.spawn(f)
+                        .map_err(move |e| warn!(log_clone, "failed to reply {:?}: {:?}", req, e));
+                    return ctx.spawn(f);
+                }
+            };
+
+        let validator_proposers: Result<Vec<usize>, _> = epoch
+            .slot_iter(spec.slots_per_epoch)
+            .map(|slot| state.get_beacon_proposer_index(slot, relative_epoch, &spec))
+            .collect();
+        let validator_proposers = match validator_proposers {
+            Ok(v) => v,
+            Err(e) => {
+                // could not get the validator proposer index
+                let log_clone = self.log.clone();
+                let f = sink
+                    .fail(RpcStatus::new(
+                        RpcStatusCode::FailedPrecondition,
+                        Some(format!("Could not find beacon proposers: {:?}", e)),
+                    ))
+                    .map_err(move |e| warn!(log_clone, "failed to reply {:?} : {:?}", req, e));
+                return ctx.spawn(f);
+            }
+        };
+
+        // get the duties for each validator
+        for validator_pk in validators.get_public_keys() {
+            let mut active_validator = ActiveValidator::new();
+
+            let public_key = match PublicKey::ssz_decode(validator_pk, 0) {
+                Ok((v, _index)) => v,
+                Err(_) => {
+                    let log_clone = self.log.clone();
+                    let f = sink
+                        .fail(RpcStatus::new(
+                            RpcStatusCode::InvalidArgument,
+                            Some("Invalid public_key".to_string()),
+                        ))
+                        .map_err(move |e| warn!(log_clone, "failed to reply {:?}", req));
+                    return ctx.spawn(f);
+                }
+            };
+
+            // get the validator index
+            let val_index = match state.get_validator_index(&public_key) {
+                Ok(Some(index)) => index,
+                Ok(None) => {
+                    // index not present in registry, set the duties for this key to None
+                    warn!(
+                        self.log,
+                        "RPC requested a public key that is not in the registry: {:?}", public_key
+                    );
+                    active_validator.set_none(false);
+                    resp_validators.push(active_validator);
+                    continue;
+                }
+                // the cache is not built, throw an error
+                Err(e) => {
+                    let log_clone = self.log.clone();
+                    let f = sink
+                        .fail(RpcStatus::new(
+                            RpcStatusCode::FailedPrecondition,
+                            Some(format!("Beacon state error {:?}", e)),
+                        ))
+                        .map_err(move |e| warn!(log_clone, "Failed to reply {:?}: {:?}", req, e));
+                    return ctx.spawn(f);
+                }
+            };
+
+            // get attestation duties and check if validator is active
+            let attestation_duties = match state.get_attestation_duties(val_index, &spec) {
+                Ok(Some(v)) => v,
+                Ok(_) => {
+                    // validator is inactive, go to the next validator
+                    warn!(
+                        self.log,
+                        "RPC requested an inactive validator key: {:?}", public_key
+                    );
+                    active_validator.set_none(false);
+                    resp_validators.push(active_validator);
+                    continue;
+                }
+                // the cache is not built, throw an error
+                Err(e) => {
+                    let log_clone = self.log.clone();
+                    let f = sink
+                        .fail(RpcStatus::new(
+                            RpcStatusCode::FailedPrecondition,
+                            Some(format!("Beacon state error {:?}", e)),
+                        ))
+                        .map_err(move |e| warn!(log_clone, "Failed to reply {:?}: {:?}", req, e));
+                    return ctx.spawn(f);
+                }
+            };
+
+            // we have an active validator, set its duties
+            let mut duty = ValidatorDuty::new();
+
+            // check if the validator needs to propose a block
+            if let Some(slot) = validator_proposers.iter().position(|&v| val_index == v) {
+                duty.set_block_production_slot(
+                    epoch.start_slot(spec.slots_per_epoch).as_u64() + slot as u64,
+                );
+            } else {
+                // no blocks to propose this epoch
+                duty.set_none(false)
+            }
+
+            duty.set_committee_index(attestation_duties.committee_index as u64);
+            duty.set_attestation_slot(attestation_duties.slot.as_u64());
+            duty.set_attestation_shard(attestation_duties.shard);
+
+            active_validator.set_duty(duty);
+            resp_validators.push(active_validator);
         }
-    }
-
-    fn propose_block_slot(
-        &mut self,
-        ctx: RpcContext,
-        req: ProposeBlockSlotRequest,
-        sink: UnarySink<ProposeBlockSlotResponse>,
-    ) {
-        debug!(self.log, "RPC request"; "endpoint" => "ProposeBlockSlot", "epoch" => req.get_epoch(), "validator_index" => req.get_validator_index());
-
-        let mut resp = ProposeBlockSlotResponse::new();
-
-        // TODO: return a legit value.
-        resp.set_slot(1);
 
         let f = sink
             .success(resp)

beacon_node/src/run.rs

@@ -6,10 +6,12 @@ use futures::Future;
 use slog::info;
 use std::cell::RefCell;
 use tokio::runtime::Builder;
+use tokio_timer::clock::Clock;
 
 pub fn run_beacon_node(config: ClientConfig, log: &slog::Logger) -> error::Result<()> {
     let mut runtime = Builder::new()
         .name_prefix("main-")
+        .clock(Clock::system())
         .build()
         .map_err(|e| format!("{:?}", e))?;
 

eth2/attester/src/lib.rs

@@ -94,7 +94,7 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> Attester<T, U, V,
     }
 
     fn produce_attestation(&mut self, slot: Slot, shard: u64) -> Result<PollOutcome, Error> {
-        let attestation_data = match self.beacon_node.produce_attestation_data(slot, shard)? {
+        let attestation_data = match self.beacon_node.produce_attestation(slot, shard)? {
             Some(attestation_data) => attestation_data,
             None => return Ok(PollOutcome::BeaconNodeUnableToProduceAttestation(slot)),
         };
@@ -119,8 +119,7 @@ impl<T: SlotClock, U: BeaconNode, V: DutiesReader, W: Signer> Attester<T, U, V,
             validator_index,
         };
 
-        self.beacon_node
+        self.beacon_node.publish_attestation(free_attestation)?;
-            .publish_attestation_data(free_attestation)?;
         Ok(PollOutcome::AttestationProduced(slot))
     }
 

eth2/attester/src/test_utils/simulated_beacon_node.rs

@@ -26,7 +26,7 @@ impl SimulatedBeaconNode {
 }
 
 impl BeaconNode for SimulatedBeaconNode {
-    fn produce_attestation_data(&self, slot: Slot, shard: u64) -> ProduceResult {
+    fn produce_attestation(&self, slot: Slot, shard: u64) -> ProduceResult {
         *self.produce_input.write().unwrap() = Some((slot, shard));
         match *self.produce_result.read().unwrap() {
             Some(ref r) => r.clone(),
@@ -34,7 +34,7 @@ impl BeaconNode for SimulatedBeaconNode {
         }
     }
 
-    fn publish_attestation_data(&self, free_attestation: FreeAttestation) -> PublishResult {
+    fn publish_attestation(&self, free_attestation: FreeAttestation) -> PublishResult {
         *self.publish_input.write().unwrap() = Some(free_attestation.clone());
         match *self.publish_result.read().unwrap() {
             Some(ref r) => r.clone(),

eth2/attester/src/traits.rs

@@ -14,13 +14,13 @@ pub enum PublishOutcome {
 
 /// Defines the methods required to produce and publish blocks on a Beacon Node.
 pub trait BeaconNode: Send + Sync {
-    fn produce_attestation_data(
+    fn produce_attestation(
         &self,
         slot: Slot,
         shard: u64,
     ) -> Result<Option<AttestationData>, BeaconNodeError>;
 
-    fn publish_attestation_data(
+    fn publish_attestation(
         &self,
         free_attestation: FreeAttestation,
     ) -> Result<PublishOutcome, BeaconNodeError>;

eth2/state_processing/src/per_slot_processing.rs

@@ -20,7 +20,6 @@ pub fn per_slot_processing(
 
     if (state.slot + 1) % spec.slots_per_epoch == 0 {
         per_epoch_processing(state, spec)?;
-        state.advance_caches();
     }
 
     state.slot += 1;

eth2/types/src/chain_spec.rs

@@ -120,7 +120,7 @@ pub struct ChainSpec {
      *
      */
     pub boot_nodes: Vec<Multiaddr>,
-    pub network_id: u8,
+    pub chain_id: u8,
 }
 
 impl ChainSpec {
@@ -257,7 +257,7 @@ impl ChainSpec {
              * Boot nodes
              */
             boot_nodes: vec![],
-            network_id: 1, // foundation network id
+            chain_id: 1, // foundation chain id
         }
     }
 
@@ -274,7 +274,7 @@ impl ChainSpec {
 
         Self {
             boot_nodes,
-            network_id: 2, // lighthouse testnet network id
+            chain_id: 2, // lighthouse testnet chain id
             ..ChainSpec::few_validators()
         }
     }

eth2/types/src/lib.rs

@@ -85,6 +85,6 @@ pub type AttesterMap = HashMap<(u64, u64), Vec<usize>>;
 pub type ProposerMap = HashMap<u64, usize>;
 
 pub use bls::{AggregatePublicKey, AggregateSignature, Keypair, PublicKey, SecretKey, Signature};
-pub use libp2p::floodsub::{Topic, TopicBuilder};
+pub use libp2p::floodsub::{Topic, TopicBuilder, TopicHash};
 pub use libp2p::multiaddr;
 pub use libp2p::Multiaddr;

eth2/types/src/test_utils/testing_beacon_state_builder.rs

@@ -120,8 +120,10 @@ impl TestingBeaconStateBuilder {
             })
             .collect();
 
+        let genesis_time = 1553753928; // arbitrary
+
         let mut state = BeaconState::genesis(
-            0,
+            genesis_time,
             Eth1Data {
                 deposit_root: Hash256::zero(),
                 block_hash: Hash256::zero(),

eth2/utils/slot_clock/src/lib.rs

@@ -3,10 +3,13 @@ mod testing_slot_clock;
 
 pub use crate::system_time_slot_clock::{Error as SystemTimeSlotClockError, SystemTimeSlotClock};
 pub use crate::testing_slot_clock::{Error as TestingSlotClockError, TestingSlotClock};
+use std::time::Duration;
 pub use types::Slot;
 
 pub trait SlotClock: Send + Sync {
     type Error;
 
     fn present_slot(&self) -> Result<Option<Slot>, Self::Error>;
+
+    fn duration_to_next_slot(&self) -> Result<Option<Duration>, Self::Error>;
 }

eth2/utils/slot_clock/src/system_time_slot_clock.rs

@@ -13,6 +13,7 @@ pub enum Error {
 /// Determines the present slot based upon the present system time.
 #[derive(Clone)]
 pub struct SystemTimeSlotClock {
+    genesis_slot: Slot,
     genesis_seconds: u64,
     slot_duration_seconds: u64,
 }
@@ -22,6 +23,7 @@ impl SystemTimeSlotClock {
     ///
     /// Returns an Error if `slot_duration_seconds == 0`.
     pub fn new(
+        genesis_slot: Slot,
         genesis_seconds: u64,
         slot_duration_seconds: u64,
     ) -> Result<SystemTimeSlotClock, Error> {
@@ -29,6 +31,7 @@ impl SystemTimeSlotClock {
             Err(Error::SlotDurationIsZero)
         } else {
             Ok(Self {
+                genesis_slot,
                 genesis_seconds,
                 slot_duration_seconds,
             })
@@ -44,11 +47,17 @@ impl SlotClock for SystemTimeSlotClock {
         let duration_since_epoch = syslot_time.duration_since(SystemTime::UNIX_EPOCH)?;
         let duration_since_genesis =
             duration_since_epoch.checked_sub(Duration::from_secs(self.genesis_seconds));
+
         match duration_since_genesis {
             None => Ok(None),
-            Some(d) => Ok(slot_from_duration(self.slot_duration_seconds, d)),
+            Some(d) => Ok(slot_from_duration(self.slot_duration_seconds, d)
+                .and_then(|s| Some(s + self.genesis_slot))),
         }
     }
+
+    fn duration_to_next_slot(&self) -> Result<Option<Duration>, Error> {
+        duration_to_next_slot(self.genesis_seconds, self.slot_duration_seconds)
+    }
 }
 
 impl From<SystemTimeError> for Error {
@@ -62,6 +71,30 @@ fn slot_from_duration(slot_duration_seconds: u64, duration: Duration) -> Option<
         duration.as_secs().checked_div(slot_duration_seconds)?,
     ))
 }
+// calculate the duration to the next slot
+fn duration_to_next_slot(
+    genesis_time: u64,
+    seconds_per_slot: u64,
+) -> Result<Option<Duration>, Error> {
+    let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
+    let genesis_time = Duration::from_secs(genesis_time);
+
+    if now < genesis_time {
+        return Ok(None);
+    }
+
+    let since_genesis = now - genesis_time;
+
+    let elapsed_slots = since_genesis.as_secs() / seconds_per_slot;
+
+    let next_slot_start_seconds = (elapsed_slots + 1)
+        .checked_mul(seconds_per_slot)
+        .expect("Next slot time should not overflow u64");
+
+    let time_to_next_slot = Duration::from_secs(next_slot_start_seconds) - since_genesis;
+
+    Ok(Some(time_to_next_slot))
+}
 
 #[cfg(test)]
 mod tests {
@@ -74,6 +107,7 @@ mod tests {
     #[test]
     fn test_slot_now() {
         let slot_time = 100;
+        let genesis_slot = Slot::new(0);
 
         let now = SystemTime::now();
         let since_epoch = now.duration_since(SystemTime::UNIX_EPOCH).unwrap();
@@ -81,18 +115,21 @@ mod tests {
         let genesis = since_epoch.as_secs() - slot_time * 89;
 
         let clock = SystemTimeSlotClock {
+            genesis_slot,
             genesis_seconds: genesis,
             slot_duration_seconds: slot_time,
         };
         assert_eq!(clock.present_slot().unwrap(), Some(Slot::new(89)));
 
         let clock = SystemTimeSlotClock {
+            genesis_slot,
             genesis_seconds: since_epoch.as_secs(),
             slot_duration_seconds: slot_time,
         };
         assert_eq!(clock.present_slot().unwrap(), Some(Slot::new(0)));
 
         let clock = SystemTimeSlotClock {
+            genesis_slot,
             genesis_seconds: since_epoch.as_secs() - slot_time * 42 - 5,
             slot_duration_seconds: slot_time,
         };

eth2/utils/slot_clock/src/testing_slot_clock.rs

@@ -1,5 +1,6 @@
 use super::SlotClock;
 use std::sync::RwLock;
+use std::time::Duration;
 use types::Slot;
 
 #[derive(Debug, PartialEq)]
@@ -32,6 +33,11 @@ impl SlotClock for TestingSlotClock {
         let slot = *self.slot.read().expect("TestingSlotClock poisoned.");
         Ok(Some(Slot::new(slot)))
     }
+
+    /// Always returns a duration of 1 second.
+    fn duration_to_next_slot(&self) -> Result<Option<Duration>, Error> {
+        Ok(Some(Duration::from_secs(1)))
+    }
 }
 
 #[cfg(test)]

protos/src/services.proto

@@ -12,24 +12,55 @@ syntax = "proto3";
 
 package ethereum.beacon.rpc.v1;
 
+// Service that currently identifies a beacon node
+service BeaconNodeService {
+    rpc Info(Empty) returns (NodeInfoResponse);
+}
+
+/// Service that handles block production
 service BeaconBlockService {
     rpc ProduceBeaconBlock(ProduceBeaconBlockRequest) returns (ProduceBeaconBlockResponse);
     rpc PublishBeaconBlock(PublishBeaconBlockRequest) returns (PublishBeaconBlockResponse);
 }
 
+/// Service that provides the validator client with requisite knowledge about
+//its public keys
 service ValidatorService {
-    // rpc ValidatorAssignment(ValidatorAssignmentRequest) returns (ValidatorAssignmentResponse);
+    // Gets the block proposer slot and comittee slot that a validator needs to
-	rpc ProposeBlockSlot(ProposeBlockSlotRequest) returns (ProposeBlockSlotResponse);
+    // perform work on.
-	rpc ValidatorIndex(PublicKey) returns (IndexResponse);
+	rpc GetValidatorDuties(GetDutiesRequest) returns (GetDutiesResponse);
 }
 
-message BeaconBlock {
+/// Service that handles validator attestations
-	uint64 slot = 1;
+service AttestationService {
-	bytes block_root = 2;
+	rpc ProduceAttestation(ProduceAttestationRequest) returns (ProduceAttestationResponse);
-	bytes randao_reveal = 3;
+	rpc PublishAttestation(PublishAttestationRequest) returns (PublishAttestationResponse);
-	bytes signature = 4;
 }
 
+/*
+ * Beacon Node Service Message
+ */
+message NodeInfoResponse {
+    string version = 1;
+    Fork fork = 2;
+    uint32 chain_id = 3;
+    uint64 genesis_time = 4;
+    uint64 genesis_slot = 5;
+}
+
+message Fork {
+    bytes previous_version = 1;
+    bytes current_version = 2;
+    uint64 epoch = 3;
+}
+
+message Empty {}
+
+
+/*
+ * Block Production Service Messages
+ */
+
 // Validator requests an unsigned proposal.
 message ProduceBeaconBlockRequest {
     uint64 slot = 1;
@@ -51,44 +82,90 @@ message PublishBeaconBlockResponse {
     bytes msg = 2;
 }
 
-// A validators duties for some epoch.
+message BeaconBlock {
-// TODO: add shard duties.
+	bytes ssz = 1;
-message ValidatorAssignment {
-	oneof block_production_slot_oneof {
-		bool block_production_slot_none = 1;
-		uint64 block_production_slot = 2;
-	}
-}
-
-message ValidatorAssignmentRequest {
-	uint64 epoch = 1;
-	bytes validator_index = 2;
 }
 
 /*
- * Propose slot
+ * Validator Service Messages
  */
 
-message ProposeBlockSlotRequest {
+// Validator Assignment
-	uint64 epoch = 1;
+
-	uint64 validator_index = 2;
+// the public keys of the validators
+message Validators {
+	repeated bytes public_keys = 1;
 }
 
-message ProposeBlockSlotResponse {
+// Propose slot
-	oneof slot_oneof {
+message GetDutiesRequest {
+	uint64 epoch = 1;
+	Validators validators = 2;
+}
+
+message GetDutiesResponse {
+	repeated ActiveValidator active_validators = 1;
+}
+
+message ActiveValidator {
+    oneof duty_oneof {
 		bool none = 1;
-		uint64 slot = 2;
+        ValidatorDuty duty = 2;
 	}
 }
 
+message ValidatorDuty {
+    oneof block_oneof {
+        bool none = 1;
+        uint64 block_production_slot = 2;
+    }
+	uint64 attestation_slot = 3;
+	uint64 attestation_shard = 4;
+    uint64 committee_index = 5;
+}
+
 /*
- * Validator Assignment
+ * Attestation Service Messages
  */
 
-message PublicKey {
+message ProduceAttestationRequest {
-	bytes public_key = 1;
+	uint64 slot = 1;
+	uint64 shard = 2;
 }
 
-message IndexResponse {
+message ProduceAttestationResponse {
-	uint64 index = 1;
+	Attestation attestation_data = 1;
+}
+
+message PublishAttestationRequest {
+	FreeAttestation free_attestation = 1;
+}
+
+message PublishAttestationResponse {
+	bool success = 1;
+	bytes msg = 2;
+}
+
+message Crosslink {
+	uint64 epoch = 1;
+	bytes crosslink_data_root = 2;
+
+}
+
+message Attestation {
+	uint64 slot = 1;
+	uint64 shard = 2;
+	bytes beacon_block_root = 3;
+	bytes epoch_boundary_root = 4;
+	bytes crosslink_data_root = 5;
+	Crosslink latest_crosslink = 6;
+	uint64 justified_epoch = 7;
+	bytes justified_block_root = 8;
+
+}
+
+message FreeAttestation {
+	Attestation attestation_data = 1;
+	bytes signature = 2;
+	uint64 validator_index = 3;
 }

validator_client/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "validator_client"
 version = "0.1.0"
-authors = ["Paul Hauner <paul@paulhauner.com>"]
+authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com>", "Luke Anderson <luke@lukeanderson.com.au>"]
 edition = "2018"
 
 [[bin]]
@@ -15,7 +15,9 @@ path = "src/lib.rs"
 
 [dependencies]
 block_proposer = { path = "../eth2/block_proposer" }
+attester = { path = "../eth2/attester" }
 bls = { path = "../eth2/utils/bls" }
+ssz = { path = "../eth2/utils/ssz" }
 clap = "2.32.0"
 dirs = "1.0.3"
 grpcio = { version = "0.4", default-features = false, features = ["protobuf-codec"] }
@@ -26,5 +28,8 @@ types = { path = "../eth2/types" }
 slog = "^2.2.3"
 slog-term = "^2.4.0"
 slog-async = "^2.3.0"
-ssz = { path = "../eth2/utils/ssz" }
+tokio = "0.1.18"
+tokio-timer = "0.2.10"
+error-chain = "0.12.0"
 bincode = "^1.1.2"
+futures = "0.1.25"

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::ProduceAttestationRequest;
+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(
+        &self,
+        slot: Slot,
+        shard: u64,
+    ) -> Result<Option<AttestationData>, BeaconNodeError> {
+        let mut req = ProduceAttestationRequest::new();
+        req.set_slot(slot.as_u64());
+        req.set_shard(shard);
+
+        let reply = self
+            .client
+            .produce_attestation(&req)
+            .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
+
+        // TODO: return correct Attestation
+        Err(BeaconNodeError::DecodeFailure)
+    }
+
+    fn publish_attestation(
+        &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/block_producer_service/beacon_block_grpc_client.rs

@@ -5,7 +5,7 @@ use protos::services::{
 use protos::services_grpc::BeaconBlockServiceClient;
 use ssz::{ssz_encode, Decodable};
 use std::sync::Arc;
-use types::{BeaconBlock, BeaconBlockBody, Eth1Data, Hash256, Signature, Slot};
+use types::{BeaconBlock, Signature, Slot};
 
 /// A newtype designed to wrap the gRPC-generated service so the `BeaconNode` trait may be
 /// implemented upon it.
@@ -40,33 +40,12 @@ impl BeaconNode for BeaconBlockGrpcClient {
 
         if reply.has_block() {
             let block = reply.get_block();
+            let ssz = block.get_ssz();
 
-            let (signature, _) = Signature::ssz_decode(block.get_signature(), 0)
+            let (block, _i) =
-                .map_err(|_| BeaconNodeError::DecodeFailure)?;
+                BeaconBlock::ssz_decode(&ssz, 0).map_err(|_| BeaconNodeError::DecodeFailure)?;
 
-            let (randao_reveal, _) = Signature::ssz_decode(block.get_randao_reveal(), 0)
+            Ok(Some(block))
-                .map_err(|_| BeaconNodeError::DecodeFailure)?;
-
-            // TODO: this conversion is incomplete; fix it.
-            Ok(Some(BeaconBlock {
-                slot: Slot::new(block.get_slot()),
-                previous_block_root: Hash256::zero(),
-                state_root: Hash256::zero(),
-                signature,
-                body: BeaconBlockBody {
-                    randao_reveal,
-                    eth1_data: Eth1Data {
-                        deposit_root: Hash256::zero(),
-                        block_hash: Hash256::zero(),
-                    },
-                    proposer_slashings: vec![],
-                    attester_slashings: vec![],
-                    attestations: vec![],
-                    deposits: vec![],
-                    voluntary_exits: vec![],
-                    transfers: vec![],
-                },
-            }))
         } else {
             Ok(None)
         }
@@ -79,12 +58,11 @@ impl BeaconNode for BeaconBlockGrpcClient {
     fn publish_beacon_block(&self, block: BeaconBlock) -> Result<PublishOutcome, BeaconNodeError> {
         let mut req = PublishBeaconBlockRequest::new();
 
+        let ssz = ssz_encode(&block);
+
         // TODO: this conversion is incomplete; fix it.
         let mut grpc_block = GrpcBeaconBlock::new();
-        grpc_block.set_slot(block.slot.as_u64());
+        grpc_block.set_ssz(ssz);
-        grpc_block.set_block_root(vec![0]);
-        grpc_block.set_randao_reveal(ssz_encode(&block.body.randao_reveal));
-        grpc_block.set_signature(ssz_encode(&block.signature));
 
         req.set_block(grpc_block);
 

validator_client/src/config.rs

@@ -22,15 +22,14 @@ pub struct Config {
 const DEFAULT_PRIVATE_KEY_FILENAME: &str = "private.key";
 
 impl Default for Config {
+    /// Build a new configuration from defaults.
     fn default() -> Self {
         let data_dir = {
             let home = dirs::home_dir().expect("Unable to determine home directory.");
             home.join(".lighthouse-validator")
         };
-        fs::create_dir_all(&data_dir)
-            .unwrap_or_else(|_| panic!("Unable to create {:?}", &data_dir));
 
-        let server = "localhost:50051".to_string();
+        let server = "localhost:5051".to_string();
 
         let spec = ChainSpec::foundation();
 
@@ -50,13 +49,14 @@ impl Config {
         // Use the specified datadir, or default in the home directory
         if let Some(datadir) = args.value_of("datadir") {
             config.data_dir = PathBuf::from(datadir);
-            fs::create_dir_all(&config.data_dir)
-                .unwrap_or_else(|_| panic!("Unable to create {:?}", &config.data_dir));
             info!(log, "Using custom data dir: {:?}", &config.data_dir);
         };
 
+        fs::create_dir_all(&config.data_dir)
+            .unwrap_or_else(|_| panic!("Unable to create {:?}", &config.data_dir));
+
         if let Some(srv) = args.value_of("server") {
-            //TODO: I don't think this parses correctly a server & port combo
+            //TODO: Validate the server value, to ensure it makes sense.
             config.server = srv.to_string();
             info!(log, "Using custom server: {:?}", &config.server);
         };
@@ -67,10 +67,15 @@ impl Config {
             config.spec = match spec_str {
                 "foundation" => ChainSpec::foundation(),
                 "few_validators" => ChainSpec::few_validators(),
+                "lighthouse_testnet" => ChainSpec::lighthouse_testnet(),
                 // Should be impossible due to clap's `possible_values(..)` function.
                 _ => unreachable!(),
             };
         };
+        // Log configuration
+        info!(log, "";
+              "data_dir" => &config.data_dir.to_str(),
+              "server" => &config.server);
 
         Ok(config)
     }

validator_client/src/duties/epoch_duties.rs

@@ -1,90 +1,123 @@
-use block_proposer::{DutiesReader, DutiesReaderError};
 use std::collections::HashMap;
-use std::sync::RwLock;
+use std::ops::{Deref, DerefMut};
-use types::{Epoch, Fork, Slot};
+use types::{AttestationDuty, Epoch, PublicKey, Slot};
+
+/// When work needs to be performed by a validator, this type is given back to the main service
+/// which indicates all the information that required to process the work.
+///
+/// Note: This is calculated per slot, so a validator knows which slot is related to this struct.
+#[derive(Debug, Clone)]
+pub struct WorkInfo {
+    /// Validator needs to produce a block.
+    pub produce_block: bool,
+    /// Validator needs to produce an attestation. This supplies the required attestation data.
+    pub attestation_duty: Option<AttestationDuty>,
+}
 
 /// 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
+/// (their 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 struct EpochDuty {
-    pub validator_index: u64,
     pub block_production_slot: Option<Slot>,
-    // Future shard info
+    pub attestation_slot: Slot,
+    pub attestation_shard: u64,
+    pub committee_index: u64,
 }
 
-impl EpochDuties {
+impl EpochDuty {
-    /// Returns `true` if the supplied `slot` is a slot in which the validator should produce a
+    /// Returns `WorkInfo` if work needs to be done in the supplied `slot`
-    /// block.
+    pub fn is_work_slot(&self, slot: Slot) -> Option<WorkInfo> {
-    pub fn is_block_production_slot(&self, slot: Slot) -> bool {
+        // if validator is required to produce a slot return true
-        match self.block_production_slot {
+        let produce_block = match self.block_production_slot {
             Some(s) if s == slot => true,
             _ => false,
+        };
+
+        // if the validator is required to attest to a shard, create the data
+        let mut attestation_duty = None;
+        if self.attestation_slot == slot {
+            attestation_duty = Some(AttestationDuty {
+                slot,
+                shard: self.attestation_shard,
+                committee_index: self.committee_index as usize,
+            });
         }
+
+        if produce_block | attestation_duty.is_some() {
+            return Some(WorkInfo {
+                produce_block,
+                attestation_duty,
+            });
+        }
+        None
     }
 }
+/// Maps a list of public keys (many validators) to an EpochDuty.
+pub type EpochDuties = HashMap<PublicKey, Option<EpochDuty>>;
 
 pub enum EpochDutiesMapError {
     Poisoned,
+    UnknownEpoch,
+    UnknownValidator,
 }
 
 /// Maps an `epoch` to some `EpochDuties` for a single validator.
 pub struct EpochDutiesMap {
     pub slots_per_epoch: u64,
-    pub map: RwLock<HashMap<Epoch, EpochDuties>>,
+    pub map: HashMap<Epoch, EpochDuties>,
 }
 
 impl EpochDutiesMap {
     pub fn new(slots_per_epoch: u64) -> Self {
         Self {
             slots_per_epoch,
-            map: RwLock::new(HashMap::new()),
+            map: HashMap::new(),
         }
     }
-
-    pub fn get(&self, epoch: Epoch) -> Result<Option<EpochDuties>, EpochDutiesMapError> {
-        let map = self.map.read().map_err(|_| EpochDutiesMapError::Poisoned)?;
-        match map.get(&epoch) {
-            Some(duties) => Ok(Some(*duties)),
-            None => Ok(None),
-        }
-    }
-
-    pub fn insert(
-        &self,
-        epoch: Epoch,
-        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 {
+// Expose the hashmap methods
-    fn is_block_production_slot(&self, slot: Slot) -> Result<bool, DutiesReaderError> {
+impl Deref for EpochDutiesMap {
+    type Target = HashMap<Epoch, EpochDuties>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.map
+    }
+}
+impl DerefMut for EpochDutiesMap {
+    fn deref_mut(&mut self) -> &mut HashMap<Epoch, EpochDuties> {
+        &mut self.map
+    }
+}
+
+impl EpochDutiesMap {
+    /// Checks if the validator has work to do.
+    pub fn is_work_slot(
+        &self,
+        slot: Slot,
+        pubkey: &PublicKey,
+    ) -> Result<Option<WorkInfo>, EpochDutiesMapError> {
         let epoch = slot.epoch(self.slots_per_epoch);
 
-        let map = self.map.read().map_err(|_| DutiesReaderError::Poisoned)?;
+        let epoch_duties = self
-        let duties = map
+            .map
             .get(&epoch)
-            .ok_or_else(|| DutiesReaderError::UnknownEpoch)?;
+            .ok_or_else(|| EpochDutiesMapError::UnknownEpoch)?;
-        Ok(duties.is_block_production_slot(slot))
+        if let Some(epoch_duty) = epoch_duties.get(pubkey) {
-    }
+            if let Some(duty) = epoch_duty {
-
+                // Retrieves the duty for a validator at a given slot
-    fn fork(&self) -> Result<Fork, DutiesReaderError> {
+                return Ok(duty.is_work_slot(slot));
-        // TODO: this is garbage data.
+            } else {
-        //
+                // the validator isn't active
-        // It will almost certainly cause signatures to fail verification.
+                return Ok(None);
-        Ok(Fork {
+            }
-            previous_version: [0; 4],
+        } else {
-            current_version: [0; 4],
+            // validator isn't known
-            epoch: Epoch::new(0),
+            return Err(EpochDutiesMapError::UnknownValidator);
-        })
+        }
     }
 }
 

validator_client/src/duties/grpc.rs

@@ -1,54 +1,60 @@
+use super::epoch_duties::{EpochDuties, EpochDuty};
 use super::traits::{BeaconNode, BeaconNodeError};
-use super::EpochDuties;
+use grpcio::CallOption;
-use protos::services::{ProposeBlockSlotRequest, PublicKey as IndexRequest};
+use protos::services::{GetDutiesRequest, Validators};
 use protos::services_grpc::ValidatorServiceClient;
 use ssz::ssz_encode;
+use std::collections::HashMap;
+use std::time::Duration;
 use types::{Epoch, PublicKey, Slot};
 
 impl BeaconNode for ValidatorServiceClient {
-    /// Request the shuffling from the Beacon Node (BN).
+    /// Requests all duties (block signing and committee attesting) from the Beacon Node (BN).
-    ///
+    fn request_duties(
-    /// As this function takes a `PublicKey`, it will first attempt to resolve the public key into
-    /// a validator index, then call the BN for production/attestation duties.
-    ///
-    /// Note: presently only block production information is returned.
-    fn request_shuffling(
         &self,
         epoch: Epoch,
-        public_key: &PublicKey,
+        pubkeys: &[PublicKey],
-    ) -> Result<Option<EpochDuties>, BeaconNodeError> {
+    ) -> Result<EpochDuties, BeaconNodeError> {
-        // Lookup the validator index for the supplied public key.
+        // Get the required duties from all validators
-        let validator_index = {
+        // build the request
-            let mut req = IndexRequest::new();
+        let mut req = GetDutiesRequest::new();
-            req.set_public_key(ssz_encode(public_key).to_vec());
-            let resp = self
-                .validator_index(&req)
-                .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
-            resp.get_index()
-        };
-
-        let mut req = ProposeBlockSlotRequest::new();
-        req.set_validator_index(validator_index);
         req.set_epoch(epoch.as_u64());
+        let mut validators = Validators::new();
+        validators.set_public_keys(pubkeys.iter().map(|v| ssz_encode(v)).collect());
+        req.set_validators(validators);
 
+        // set a timeout for requests
+        // let call_opt = CallOption::default().timeout(Duration::from_secs(2));
+
+        // send the request, get the duties reply
         let reply = self
-            .propose_block_slot(&req)
+            .get_validator_duties(&req)
             .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
 
-        let block_production_slot = if reply.has_slot() {
+        let mut epoch_duties: HashMap<PublicKey, Option<EpochDuty>> = HashMap::new();
-            Some(reply.get_slot())
+        for (index, validator_duty) in reply.get_active_validators().iter().enumerate() {
-        } else {
+            if !validator_duty.has_duty() {
-            None
+                // validator is inactive
-        };
+                epoch_duties.insert(pubkeys[index].clone(), None);
-
+                continue;
-        let block_production_slot = match block_production_slot {
+            }
-            Some(slot) => Some(Slot::new(slot)),
+            // active validator
-            None => None,
+            let active_duty = validator_duty.get_duty();
-        };
+            let block_production_slot = {
-
+                if active_duty.has_block_production_slot() {
-        Ok(Some(EpochDuties {
+                    Some(Slot::from(active_duty.get_block_production_slot()))
-            validator_index,
+                } else {
-            block_production_slot,
+                    None
-        }))
+                }
+            };
+            let epoch_duty = EpochDuty {
+                block_production_slot,
+                attestation_slot: Slot::from(active_duty.get_attestation_slot()),
+                attestation_shard: active_duty.get_attestation_shard(),
+                committee_index: active_duty.get_committee_index(),
+            };
+            epoch_duties.insert(pubkeys[index].clone(), Some(epoch_duty));
+        }
+        Ok(epoch_duties)
     }
 }

validator_client/src/duties/mod.rs

@@ -1,21 +1,21 @@
 mod epoch_duties;
 mod grpc;
-mod service;
+// TODO: reintroduce tests
-#[cfg(test)]
+//#[cfg(test)]
-mod test_node;
+//mod test_node;
 mod traits;
 
-pub use self::epoch_duties::EpochDutiesMap;
 use self::epoch_duties::{EpochDuties, EpochDutiesMapError};
-pub use self::service::DutiesManagerService;
+pub use self::epoch_duties::{EpochDutiesMap, WorkInfo};
 use self::traits::{BeaconNode, BeaconNodeError};
-use bls::PublicKey;
+use futures::Async;
-use slot_clock::SlotClock;
+use slog::{debug, error, info};
 use std::sync::Arc;
-use types::{ChainSpec, Epoch};
+use std::sync::RwLock;
+use types::{Epoch, PublicKey, Slot};
 
-#[derive(Debug, PartialEq, Clone, Copy)]
+#[derive(Debug, PartialEq, Clone)]
-pub enum PollOutcome {
+pub enum UpdateOutcome {
     /// The `EpochDuties` were not updated during this poll.
     NoChange(Epoch),
     /// The `EpochDuties` for the `epoch` were previously unknown, but obtained in the poll.
@@ -23,79 +23,117 @@ pub enum PollOutcome {
     /// New `EpochDuties` were obtained, different to those which were previously known. This is
     /// likely to be the result of chain re-organisation.
     DutiesChanged(Epoch, EpochDuties),
-    /// The Beacon Node was unable to return the duties as the validator is unknown, or the
-    /// shuffling for the epoch is unknown.
-    UnknownValidatorOrEpoch(Epoch),
 }
 
 #[derive(Debug, PartialEq)]
 pub enum Error {
-    SlotClockError,
+    DutiesMapPoisoned,
-    SlotUnknowable,
     EpochMapPoisoned,
     BeaconNodeError(BeaconNodeError),
+    UnknownEpoch,
+    UnknownValidator,
 }
 
 /// A polling state machine which ensures the latest `EpochDuties` are obtained from the Beacon
 /// Node.
 ///
-/// There is a single `DutiesManager` per validator instance.
+/// This keeps track of all validator keys and required voting slots.
-pub struct DutiesManager<T: SlotClock, U: BeaconNode> {
+pub struct DutiesManager<U: BeaconNode> {
-    pub duties_map: Arc<EpochDutiesMap>,
+    pub duties_map: RwLock<EpochDutiesMap>,
-    /// The validator's public key.
+    /// A list of all public keys known to the validator service.
-    pub pubkey: PublicKey,
+    pub pubkeys: Vec<PublicKey>,
-    pub spec: Arc<ChainSpec>,
-    pub slot_clock: Arc<T>,
     pub beacon_node: Arc<U>,
 }
 
-impl<T: SlotClock, U: BeaconNode> DutiesManager<T, U> {
+impl<U: BeaconNode> DutiesManager<U> {
-    /// Poll the Beacon Node for `EpochDuties`.
+    /// Check the Beacon Node for `EpochDuties`.
     ///
-    /// The present `epoch` will be learned from the supplied `SlotClock`. In production this will
     /// be a wall-clock (e.g., system time, remote server time, etc.).
-    pub fn poll(&self) -> Result<PollOutcome, Error> {
+    fn update(&self, epoch: Epoch) -> Result<UpdateOutcome, Error> {
-        let slot = self
+        let duties = self.beacon_node.request_duties(epoch, &self.pubkeys)?;
-            .slot_clock
+        {
-            .present_slot()
-            .map_err(|_| Error::SlotClockError)?
-            .ok_or(Error::SlotUnknowable)?;
-
-        let epoch = slot.epoch(self.spec.slots_per_epoch);
-
-        if let Some(duties) = self.beacon_node.request_shuffling(epoch, &self.pubkey)? {
             // If these duties were known, check to see if they're updates or identical.
-            let result = if let Some(known_duties) = self.duties_map.get(epoch)? {
+            if let Some(known_duties) = self.duties_map.read()?.get(&epoch) {
-                if known_duties == duties {
+                if *known_duties == duties {
-                    Ok(PollOutcome::NoChange(epoch))
+                    return Ok(UpdateOutcome::NoChange(epoch));
-                } else {
-                    Ok(PollOutcome::DutiesChanged(epoch, duties))
                 }
-            } else {
+            }
-                Ok(PollOutcome::NewDuties(epoch, duties))
-            };
-            self.duties_map.insert(epoch, duties)?;
-            result
-        } else {
-            Ok(PollOutcome::UnknownValidatorOrEpoch(epoch))
         }
+        if !self.duties_map.read()?.contains_key(&epoch) {
+            //TODO: Remove clone by removing duties from outcome
+            self.duties_map.write()?.insert(epoch, duties.clone());
+            return Ok(UpdateOutcome::NewDuties(epoch, duties));
+        }
+        // duties have changed
+        //TODO: Duties could be large here. Remove from display and avoid the clone.
+        self.duties_map.write()?.insert(epoch, duties.clone());
+        return Ok(UpdateOutcome::DutiesChanged(epoch, duties));
+    }
+
+    /// A future wrapping around `update()`. This will perform logic based upon the update
+    /// process and complete once the update has completed.
+    pub fn run_update(&self, epoch: Epoch, log: slog::Logger) -> Result<Async<()>, ()> {
+        match self.update(epoch) {
+            Err(error) => error!(log, "Epoch duties poll error"; "error" => format!("{:?}", error)),
+            Ok(UpdateOutcome::NoChange(epoch)) => {
+                debug!(log, "No change in duties"; "epoch" => epoch)
+            }
+            Ok(UpdateOutcome::DutiesChanged(epoch, duties)) => {
+                info!(log, "Duties changed (potential re-org)"; "epoch" => epoch, "duties" => format!("{:?}", duties))
+            }
+            Ok(UpdateOutcome::NewDuties(epoch, duties)) => {
+                info!(log, "New duties obtained"; "epoch" => epoch, "duties" => format!("{:?}", duties))
+            }
+        };
+        Ok(Async::Ready(()))
+    }
+
+    /// Returns a list of (Public, WorkInfo) indicating all the validators that have work to perform
+    /// this slot.
+    pub fn get_current_work(&self, slot: Slot) -> Option<Vec<(PublicKey, WorkInfo)>> {
+        let mut current_work: Vec<(PublicKey, WorkInfo)> = Vec::new();
+
+        // if the map is poisoned, return None
+        let duties = self.duties_map.read().ok()?;
+
+        for validator_pk in &self.pubkeys {
+            match duties.is_work_slot(slot, &validator_pk) {
+                Ok(Some(work_type)) => current_work.push((validator_pk.clone(), work_type)),
+                Ok(None) => {} // No work for this validator
+                Err(_) => {}   // Unknown epoch or validator, no work
+            }
+        }
+        if current_work.is_empty() {
+            return None;
+        }
+        Some(current_work)
     }
 }
 
+//TODO: Use error_chain to handle errors
 impl From<BeaconNodeError> for Error {
     fn from(e: BeaconNodeError) -> Error {
         Error::BeaconNodeError(e)
     }
 }
 
+//TODO: Use error_chain to handle errors
+impl<T> From<std::sync::PoisonError<T>> for Error {
+    fn from(_e: std::sync::PoisonError<T>) -> Error {
+        Error::DutiesMapPoisoned
+    }
+}
 impl From<EpochDutiesMapError> for Error {
     fn from(e: EpochDutiesMapError) -> Error {
         match e {
             EpochDutiesMapError::Poisoned => Error::EpochMapPoisoned,
+            EpochDutiesMapError::UnknownEpoch => Error::UnknownEpoch,
+            EpochDutiesMapError::UnknownValidator => Error::UnknownValidator,
         }
     }
 }
 
+/* TODO: Modify tests for new Duties Manager form
 #[cfg(test)]
 mod tests {
     use super::test_node::TestBeaconNode;
@@ -109,6 +147,7 @@ mod tests {
     //
     // These tests should serve as a good example for future tests.
 
+
     #[test]
     pub fn polling() {
         let spec = Arc::new(ChainSpec::foundation());
@@ -159,3 +198,4 @@ mod tests {
         );
     }
 }
+*/

validator_client/src/duties/service.rs

@@ -1,40 +0,0 @@
-use super::traits::BeaconNode;
-use super::{DutiesManager, PollOutcome};
-use slog::{debug, error, info, Logger};
-use slot_clock::SlotClock;
-use std::time::Duration;
-
-pub struct DutiesManagerService<T: SlotClock, U: BeaconNode> {
-    pub manager: DutiesManager<T, U>,
-    pub poll_interval_millis: u64,
-    pub log: Logger,
-}
-
-impl<T: SlotClock, U: BeaconNode> DutiesManagerService<T, U> {
-    /// Run a loop which polls the manager each `poll_interval_millis` milliseconds.
-    ///
-    /// Logs the results of the polls.
-    pub fn run(&mut self) {
-        loop {
-            match self.manager.poll() {
-                Err(error) => {
-                    error!(self.log, "Epoch duties poll error"; "error" => format!("{:?}", error))
-                }
-                Ok(PollOutcome::NoChange(epoch)) => {
-                    debug!(self.log, "No change in duties"; "epoch" => epoch)
-                }
-                Ok(PollOutcome::DutiesChanged(epoch, duties)) => {
-                    info!(self.log, "Duties changed (potential re-org)"; "epoch" => epoch, "duties" => format!("{:?}", duties))
-                }
-                Ok(PollOutcome::NewDuties(epoch, duties)) => {
-                    info!(self.log, "New duties obtained"; "epoch" => epoch, "duties" => format!("{:?}", duties))
-                }
-                Ok(PollOutcome::UnknownValidatorOrEpoch(epoch)) => {
-                    error!(self.log, "Epoch or validator unknown"; "epoch" => epoch)
-                }
-            };
-
-            std::thread::sleep(Duration::from_millis(self.poll_interval_millis));
-        }
-    }
-}

validator_client/src/duties/traits.rs

@@ -1,6 +1,5 @@
 use super::EpochDuties;
-use bls::PublicKey;
+use types::{Epoch, PublicKey};
-use types::Epoch;
 
 #[derive(Debug, PartialEq, Clone)]
 pub enum BeaconNodeError {
@@ -9,12 +8,13 @@ pub enum BeaconNodeError {
 
 /// Defines the methods required to obtain a validators shuffling from a Beacon Node.
 pub trait BeaconNode: Send + Sync {
-    /// Get the shuffling for the given epoch and public key.
+    /// Gets the duties for all validators.
     ///
-    /// Returns Ok(None) if the public key is unknown, or the shuffling for that epoch is unknown.
+    /// Returns a vector of EpochDuties for each validator public key. The entry will be None for
-    fn request_shuffling(
+    /// validators that are not activated.
+    fn request_duties(
         &self,
         epoch: Epoch,
-        public_key: &PublicKey,
+        pubkeys: &[PublicKey],
-    ) -> Result<Option<EpochDuties>, BeaconNodeError>;
+    ) -> Result<EpochDuties, BeaconNodeError>;
 }

validator_client/src/error.rs

@@ -0,0 +1,22 @@
+use slot_clock;
+
+use error_chain::{
+    error_chain, error_chain_processing, impl_error_chain_kind, impl_error_chain_processed,
+    impl_extract_backtrace,
+};
+
+error_chain! {
+   links  { }
+
+   errors {
+    SlotClockError(e: slot_clock::SystemTimeSlotClockError) {
+        description("Error reading system time"),
+        display("SlotClockError: '{:?}'", e)
+    }
+
+    SystemTimeError(t: String ) {
+        description("Error reading system time"),
+        display("SystemTimeError: '{}'", t)
+    }
+   }
+}

validator_client/src/main.rs

@@ -1,18 +1,14 @@
-use self::block_producer_service::{BeaconBlockGrpcClient, BlockProducerService};
+mod attester_service;
-use self::duties::{DutiesManager, DutiesManagerService, EpochDutiesMap};
-use crate::config::Config;
-use block_proposer::{test_utils::LocalSigner, BlockProducer};
-use clap::{App, Arg};
-use grpcio::{ChannelBuilder, EnvBuilder};
-use protos::services_grpc::{BeaconBlockServiceClient, ValidatorServiceClient};
-use slog::{info, o, Drain};
-use slot_clock::SystemTimeSlotClock;
-use std::sync::Arc;
-use std::thread;
-
 mod block_producer_service;
 mod config;
 mod duties;
+pub mod error;
+mod service;
+
+use crate::config::Config as ValidatorClientConfig;
+use clap::{App, Arg};
+use service::Service as ValidatorService;
+use slog::{error, info, o, Drain};
 
 fn main() {
     // Logging
@@ -47,116 +43,17 @@ fn main() {
                 .short("s")
                 .help("Configuration of Beacon Chain")
                 .takes_value(true)
-                .possible_values(&["foundation", "few_validators"])
+                .possible_values(&["foundation", "few_validators", "lighthouse_testnet"])
-                .default_value("foundation"),
+                .default_value("lighthouse_testnet"),
         )
         .get_matches();
 
-    let config = Config::parse_args(&matches, &log)
+    let config = ValidatorClientConfig::parse_args(&matches, &log)
         .expect("Unable to build a configuration for the validator client.");
 
-    // Log configuration
+    // start the validator service.
-    info!(log, "Configuration parameters:";
+    match ValidatorService::start(config, log.clone()) {
-          "data_dir" => &config.data_dir.to_str(),
+        Ok(_) => info!(log, "Validator client shutdown successfully."),
-          "server" => &config.server);
+        Err(e) => error!(log, "Validator exited due to: {}", e.to_string()),
-
-    // Beacon node gRPC beacon block endpoints.
-    let beacon_block_grpc_client = {
-        let env = Arc::new(EnvBuilder::new().build());
-        let ch = ChannelBuilder::new(env).connect(&config.server);
-        Arc::new(BeaconBlockServiceClient::new(ch))
-    };
-
-    // Beacon node gRPC validator endpoints.
-    let validator_grpc_client = {
-        let env = Arc::new(EnvBuilder::new().build());
-        let ch = ChannelBuilder::new(env).connect(&config.server);
-        Arc::new(ValidatorServiceClient::new(ch))
-    };
-
-    // Spec
-    let spec = Arc::new(config.spec.clone());
-
-    // Clock for determining the present slot.
-    // TODO: this shouldn't be a static time, instead it should be pulled from the beacon node.
-    // https://github.com/sigp/lighthouse/issues/160
-    let genesis_time = 1_549_935_547;
-    let slot_clock = {
-        info!(log, "Genesis time"; "unix_epoch_seconds" => genesis_time);
-        let clock = SystemTimeSlotClock::new(genesis_time, spec.seconds_per_slot)
-            .expect("Unable to instantiate SystemTimeSlotClock.");
-        Arc::new(clock)
-    };
-
-    let poll_interval_millis = spec.seconds_per_slot * 1000 / 10; // 10% epoch time precision.
-    info!(log, "Starting block producer service"; "polls_per_epoch" => spec.seconds_per_slot * 1000 / poll_interval_millis);
-
-    let keypairs = config.fetch_keys(&log)
-        .expect("No key pairs found in configuration, they must first be generated with: account_manager generate.");
-
-    /*
-     * Start threads.
-     */
-    let mut threads = vec![];
-
-    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));
-
-        // Spawn a new thread to maintain the validator's `EpochDuties`.
-        let duties_manager_thread = {
-            let spec = spec.clone();
-            let duties_map = duties_map.clone();
-            let slot_clock = slot_clock.clone();
-            let log = log.clone();
-            let beacon_node = validator_grpc_client.clone();
-            let pubkey = keypair.pk.clone();
-            thread::spawn(move || {
-                let manager = DutiesManager {
-                    duties_map,
-                    pubkey,
-                    spec,
-                    slot_clock,
-                    beacon_node,
-                };
-                let mut duties_manager_service = DutiesManagerService {
-                    manager,
-                    poll_interval_millis,
-                    log,
-                };
-
-                duties_manager_service.run();
-            })
-        };
-
-        // 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 duties_map = duties_map.clone();
-            let slot_clock = slot_clock.clone();
-            let log = log.clone();
-            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);
-                let mut block_producer_service = BlockProducerService {
-                    block_producer,
-                    poll_interval_millis,
-                    log,
-                };
-
-                block_producer_service.run();
-            })
-        };
-
-        threads.push((duties_manager_thread, producer_thread));
-    }
-
-    // Naively wait for all the threads to complete.
-    for tuple in threads {
-        let (manager, producer) = tuple;
-        let _ = producer.join();
-        let _ = manager.join();
     }
 }

validator_client/src/service.rs

@@ -0,0 +1,328 @@
+/// The validator service. Connects to a beacon node and signs blocks when required.
+use crate::attester_service::{AttestationGrpcClient, AttesterService};
+use crate::block_producer_service::{BeaconBlockGrpcClient, BlockProducerService};
+use crate::config::Config as ValidatorConfig;
+use crate::duties::UpdateOutcome;
+use crate::duties::{DutiesManager, EpochDutiesMap};
+use crate::error as error_chain;
+use crate::error::ErrorKind;
+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 grpcio::{ChannelBuilder, EnvBuilder};
+use protos::services::Empty;
+use protos::services_grpc::{
+    AttestationServiceClient, BeaconBlockServiceClient, BeaconNodeServiceClient,
+    ValidatorServiceClient,
+};
+use slog::{debug, error, info, warn};
+use slot_clock::{SlotClock, SystemTimeSlotClock};
+use std::sync::Arc;
+use std::sync::RwLock;
+use std::time::{Duration, Instant, SystemTime};
+use tokio::prelude::*;
+use tokio::runtime::Builder;
+use tokio::timer::Interval;
+use tokio_timer::clock::Clock;
+use types::test_utils::generate_deterministic_keypairs;
+use types::{Epoch, Fork, Slot};
+
+//TODO: This service should be simplified in the future. Can be made more steamlined.
+
+/// The validator service. This is the main thread that executes and maintains validator
+/// duties.
+pub struct Service {
+    /// The node we currently connected to.
+    connected_node_version: String,
+    /// The chain id we are processing on.
+    chain_id: u16,
+    /// The fork state we processing on.
+    fork: Fork,
+    /// The slot clock for this service.
+    slot_clock: SystemTimeSlotClock,
+    /// The current slot we are processing.
+    current_slot: Slot,
+    /// The number of slots per epoch to allow for converting slots to epochs.
+    slots_per_epoch: u64,
+    // GRPC Clients
+    /// The beacon block GRPC client.
+    beacon_block_client: Arc<BeaconBlockServiceClient>,
+    /// The validator GRPC client.
+    validator_client: Arc<ValidatorServiceClient>,
+    /// The attester GRPC client.
+    attester_client: Arc<AttestationServiceClient>,
+    /// The validator client logger.
+    log: slog::Logger,
+}
+
+impl Service {
+    ///  Initial connection to the beacon node to determine its properties.
+    ///
+    ///  This tries to connect to a beacon node. Once connected, it initialised the gRPC clients
+    ///  and returns an instance of the service.
+    fn initialize_service(
+        config: &ValidatorConfig,
+        log: slog::Logger,
+    ) -> error_chain::Result<Self> {
+        // initialise the beacon node client to check for a connection
+
+        let env = Arc::new(EnvBuilder::new().build());
+        // Beacon node gRPC beacon node endpoints.
+        let beacon_node_client = {
+            let ch = ChannelBuilder::new(env.clone()).connect(&config.server);
+            Arc::new(BeaconNodeServiceClient::new(ch))
+        };
+
+        // retrieve node information
+        let node_info = loop {
+            match beacon_node_client.info(&Empty::new()) {
+                Err(e) => {
+                    warn!(log, "Could not connect to node. Error: {}", e);
+                    info!(log, "Retrying in 5 seconds...");
+                    std::thread::sleep(Duration::from_secs(5));
+                    continue;
+                }
+                Ok(info) => {
+                    if SystemTime::now()
+                        .duration_since(SystemTime::UNIX_EPOCH)
+                        .unwrap()
+                        .as_secs()
+                        < info.genesis_time
+                    {
+                        warn!(
+                            log,
+                            "Beacon Node's genesis time is in the future. No work to do.\n Exiting"
+                        );
+                        return Err("Genesis time in the future".into());
+                    }
+                    break info;
+                }
+            };
+        };
+
+        // build requisite objects to form Self
+        let genesis_time = node_info.get_genesis_time();
+        let genesis_slot = Slot::from(node_info.get_genesis_slot());
+
+        info!(log,"Beacon node connected"; "Node Version" => node_info.version.clone(), "Chain ID" => node_info.chain_id, "Genesis time" => genesis_time);
+
+        let proto_fork = node_info.get_fork();
+        let mut previous_version: [u8; 4] = [0; 4];
+        let mut current_version: [u8; 4] = [0; 4];
+        previous_version.copy_from_slice(&proto_fork.get_previous_version()[..4]);
+        current_version.copy_from_slice(&proto_fork.get_current_version()[..4]);
+        let fork = Fork {
+            previous_version,
+            current_version,
+            epoch: Epoch::from(proto_fork.get_epoch()),
+        };
+
+        // initialize the RPC clients
+
+        // Beacon node gRPC beacon block endpoints.
+        let beacon_block_client = {
+            let ch = ChannelBuilder::new(env.clone()).connect(&config.server);
+            Arc::new(BeaconBlockServiceClient::new(ch))
+        };
+
+        // Beacon node gRPC validator endpoints.
+        let validator_client = {
+            let ch = ChannelBuilder::new(env.clone()).connect(&config.server);
+            Arc::new(ValidatorServiceClient::new(ch))
+        };
+
+        //Beacon node gRPC attester endpoints.
+        let attester_client = {
+            let ch = ChannelBuilder::new(env.clone()).connect(&config.server);
+            Arc::new(AttestationServiceClient::new(ch))
+        };
+
+        // build the validator slot clock
+        let slot_clock =
+            SystemTimeSlotClock::new(genesis_slot, genesis_time, config.spec.seconds_per_slot)
+                .expect("Unable to instantiate SystemTimeSlotClock.");
+
+        let current_slot = slot_clock
+            .present_slot()
+            .map_err(|e| ErrorKind::SlotClockError(e))?
+            .expect("Genesis must be in the future");
+
+        Ok(Self {
+            connected_node_version: node_info.version,
+            chain_id: node_info.chain_id as u16,
+            fork,
+            slot_clock,
+            current_slot,
+            slots_per_epoch: config.spec.slots_per_epoch,
+            beacon_block_client,
+            validator_client,
+            attester_client,
+            log,
+        })
+    }
+
+    /// Initialise the service then run the core thread.
+    pub fn start(config: ValidatorConfig, log: slog::Logger) -> error_chain::Result<()> {
+        // connect to the node and retrieve its properties and initialize the gRPC clients
+        let service = Service::initialize_service(&config, log)?;
+
+        // we have connected to a node and established its parameters. Spin up the core service
+
+        // set up the validator service runtime
+        let mut runtime = Builder::new()
+            .clock(Clock::system())
+            .name_prefix("validator-client-")
+            .build()
+            .map_err(|e| format!("Tokio runtime failed: {}", e))?;
+
+        let duration_to_next_slot = service
+            .slot_clock
+            .duration_to_next_slot()
+            .map_err(|e| format!("System clock error: {:?}", e))?
+            .expect("Cannot start before genesis");
+
+        // set up the validator work interval - start at next slot and proceed every slot
+        let interval = {
+            // Set the interval to start at the next slot, and every slot after
+            let slot_duration = Duration::from_secs(config.spec.seconds_per_slot);
+            //TODO: Handle checked add correctly
+            Interval::new(Instant::now() + duration_to_next_slot, slot_duration)
+        };
+
+        /* kick off core service */
+
+        // generate keypairs
+
+        // TODO: keypairs are randomly generated; they should be loaded from a file or generated.
+        // https://github.com/sigp/lighthouse/issues/160
+        let keypairs = Arc::new(generate_deterministic_keypairs(8));
+
+        /* build requisite objects to pass to core thread */
+
+        // Builds a mapping of Epoch -> Map(PublicKey, EpochDuty)
+        // where EpochDuty contains slot numbers and attestation data that each validator needs to
+        // produce work on.
+        let duties_map = RwLock::new(EpochDutiesMap::new(config.spec.slots_per_epoch));
+
+        // builds a manager which maintains the list of current duties for all known validators
+        // and can check when a validator needs to perform a task.
+        let manager = Arc::new(DutiesManager {
+            duties_map,
+            pubkeys: keypairs.iter().map(|keypair| keypair.pk.clone()).collect(),
+            beacon_node: service.validator_client.clone(),
+        });
+
+        // run the core thread
+        runtime.block_on(
+            interval
+                .for_each(move |_| {
+                    let log = service.log.clone();
+
+                    /* get the current slot and epoch */
+                    let current_slot = match service.slot_clock.present_slot() {
+                        Err(e) => {
+                            error!(log, "SystemTimeError {:?}", e);
+                            return Ok(());
+                        }
+                        Ok(slot) => slot.expect("Genesis is in the future"),
+                    };
+
+                    let current_epoch = current_slot.epoch(service.slots_per_epoch);
+
+                    debug_assert!(
+                        current_slot > service.current_slot,
+                        "The Timer should poll a new slot"
+                    );
+
+                    info!(log, "Processing slot: {}", current_slot.as_u64());
+
+                    /* check for new duties */
+
+                    let cloned_manager = manager.clone();
+                    let cloned_log = log.clone();
+                    // spawn a new thread separate to the runtime
+                    std::thread::spawn(move || {
+                        cloned_manager.run_update(current_epoch.clone(), cloned_log.clone());
+                        dbg!("Finished thread");
+                    });
+
+                    /* execute any specified duties */
+
+                    if let Some(work) = manager.get_current_work(current_slot) {
+                        for (_public_key, work_type) in work {
+                            if work_type.produce_block {
+                                // TODO: Produce a beacon block in a new thread
+                            }
+                            if work_type.attestation_duty.is_some() {
+                                // available AttestationDuty info
+                                let attestation_duty =
+                                    work_type.attestation_duty.expect("Cannot be None");
+                                //TODO: Produce an attestation in a new thread
+                            }
+                        }
+                    }
+
+                    Ok(())
+                })
+                .map_err(|e| format!("Service thread failed: {:?}", e)),
+        );
+
+        // completed a slot process
+        Ok(())
+    }
+
+    /*
+
+        // Spawn a new thread to perform block production for the validator.
+        let producer_thread = {
+            let spec = spec.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();
+            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);
+                let mut block_producer_service = BlockProducerService {
+                    block_producer,
+                    poll_interval_millis,
+                    log,
+                };
+
+                block_producer_service.run();
+            })
+        };
+
+        // 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, attester) = tuple;
+        let _ = producer.join();
+        let _ = manager.join();
+        let _ = attester.join();
+    }
+    */
+}