Fixing merge conflict, having keys generated deterministically for testing.

Jenkinsfile

@@ -1,20 +1,22 @@
 pipeline {
-    agent {
+	agent {
 		dockerfile {
 			filename 'Dockerfile'
 			args '-v cargo-cache:/cargocache:rw -e "CARGO_HOME=/cargocache"'
 		}
 	}
-    stages {
+	stages {
-        stage('Build') {
+		stage('Build') {
-            steps {
-                sh 'cargo build'
-            }
-        }
-        stage('Test') {
 			steps {
-				sh 'cargo test --all'
+				sh 'cargo build --verbose --all'
+				sh 'cargo build --verbose --all --release'
 			}
 		}
-    }
+		stage('Test') {
+			steps {
+				sh 'cargo test --verbose --all'
+				sh 'cargo test --verbose --all --release'
+			}
+		}
+	}
 }

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

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/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.
@@ -26,6 +34,8 @@ pub struct Client<T: ClientTypes> {
     pub network: Arc<NetworkService>,
     /// Signal to terminate the RPC server.
     pub rpc_exit_signal: Option<Signal>,
+    /// Signal to terminate the slot timer.
+    pub slot_timer_exit_signal: Option<Signal>,
     /// The clients logger.
     log: slog::Logger,
     /// Marker to pin the beacon chain generics.
@@ -42,6 +52,35 @@ impl<TClientType: ClientTypes> Client<TClientType> {
         // 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;
@@ -65,13 +104,73 @@ impl<TClientType: ClientTypes> Client<TClientType> {
             ));
         }
 
+        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,
             rpc_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

@@ -49,12 +49,15 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
     fn inject_event(&mut self, event: GossipsubEvent) {
         match event {
             GossipsubEvent::Message(gs_msg) => {
+                debug!(self.log, "Received GossipEvent"; "msg" => format!("{:?}", gs_msg));
+
                 let pubsub_message = match PubsubMessage::ssz_decode(&gs_msg.data, 0) {
                     //TODO: Punish peer on error
                     Err(e) => {
                         warn!(
                             self.log,
-                            "Received undecodable message from Peer {:?}", gs_msg.source
+                            "Received undecodable message from Peer {:?} error", gs_msg.source;
+                            "error" => format!("{:?}", e)
                         );
                         return;
                     }
@@ -192,7 +195,7 @@ pub enum BehaviourEvent {
 }
 
 /// Messages that are passed to and from the pubsub (Gossipsub) behaviour.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum PubsubMessage {
     /// Gossipsub message providing notification of a new block.
     Block(BlockRootSlot),
@@ -220,11 +223,11 @@ impl Decodable for PubsubMessage {
     fn ssz_decode(bytes: &[u8], index: usize) -> Result<(Self, usize), DecodeError> {
         let (id, index) = u32::ssz_decode(bytes, index)?;
         match id {
-            1 => {
+            0 => {
                 let (block, index) = BlockRootSlot::ssz_decode(bytes, index)?;
                 Ok((PubsubMessage::Block(block), index))
             }
-            2 => {
+            1 => {
                 let (attestation, index) = Attestation::ssz_decode(bytes, index)?;
                 Ok((PubsubMessage::Attestation(attestation), index))
             }
@@ -232,3 +235,25 @@ impl Decodable for PubsubMessage {
         }
     }
 }
+
+#[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/network/src/sync/simple_sync.rs

@@ -523,9 +523,9 @@ impl SimpleSync {
         msg: BlockRootSlot,
         network: &mut NetworkContext,
     ) {
-        debug!(
+        info!(
             self.log,
-            "BlockSlot";
+            "NewGossipBlock";
             "peer" => format!("{:?}", peer_id),
         );
         // TODO: filter out messages that a prior to the finalized slot.
@@ -557,9 +557,9 @@ impl SimpleSync {
         msg: Attestation,
         _network: &mut NetworkContext,
     ) {
-        debug!(
+        info!(
             self.log,
-            "Attestation";
+            "NewAttestationGossip";
             "peer" => format!("{:?}", peer_id),
         );
 

beacon_node/rpc/src/beacon_block.rs

@@ -1,3 +1,4 @@
+use crate::beacon_chain::BeaconChain;
 use crossbeam_channel;
 use eth2_libp2p::rpc::methods::BlockRootSlot;
 use eth2_libp2p::PubsubMessage;
@@ -10,10 +11,14 @@ use protos::services::{
 };
 use protos::services_grpc::BeaconBlockService;
 use slog::Logger;
-use types::{Hash256, Slot};
+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,
 }
@@ -30,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);
@@ -49,26 +53,88 @@ impl BeaconBlockService for BeaconBlockServiceInstance {
         req: PublishBeaconBlockRequest,
         sink: UnarySink<PublishBeaconBlockResponse>,
     ) {
-        let block = req.get_block();
-        let block_root = Hash256::from_slice(block.get_block_root());
-        let block_slot = BlockRootSlot {
-            block_root,
-            slot: Slot::from(block.get_slot()),
-        };
-        println!("publishing block with root {:?}", block_root);
-
-        // TODO: Obtain topics from the network service properly.
-        let topic = types::TopicBuilder::new("beacon_chain".to_string()).build();
-        let message = PubsubMessage::Block(block_slot);
-        println!("Sending beacon block to gossipsub");
-        self.network_chan.send(NetworkMessage::Publish {
-            topics: vec![topic],
-            message,
-        });
-
-        // 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

@@ -5,14 +5,18 @@ use beacon_chain::{
     parking_lot::RwLockReadGuard,
     slot_clock::SlotClock,
     types::{BeaconState, ChainSpec},
-    CheckPoint,
 };
+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>
@@ -28,4 +32,11 @@ where
     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

@@ -1,7 +1,7 @@
 use crate::beacon_chain::BeaconChain;
 use futures::Future;
 use grpcio::{RpcContext, UnarySink};
-use protos::services::{Empty, Fork, NodeInfo};
+use protos::services::{Empty, Fork, NodeInfoResponse};
 use protos::services_grpc::BeaconNodeService;
 use slog::{trace, warn};
 use std::sync::Arc;
@@ -14,11 +14,11 @@ pub struct BeaconNodeServiceInstance {
 
 impl BeaconNodeService for BeaconNodeServiceInstance {
     /// Provides basic node information.
-    fn info(&mut self, ctx: RpcContext, _req: Empty, sink: UnarySink<NodeInfo>) {
+    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 = NodeInfo::new();
+        let mut node_info = NodeInfoResponse::new();
         node_info.set_version(version::version());
 
         // get the chain state
@@ -34,6 +34,7 @@ impl BeaconNodeService for BeaconNodeServiceInstance {
 
         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

beacon_node/rpc/src/lib.rs

@@ -43,6 +43,7 @@ pub fn start_server(
 
     let beacon_block_service = {
         let instance = BeaconBlockServiceInstance {
+            chain: beacon_chain.clone(),
             network_chan,
             log: log.clone(),
         };

beacon_node/rpc/src/validator.rs

@@ -4,9 +4,10 @@ use futures::Future;
 use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink};
 use protos::services::{ActiveValidator, GetDutiesRequest, GetDutiesResponse, ValidatorDuty};
 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 {
@@ -28,23 +29,47 @@ impl ValidatorService for ValidatorServiceInstance {
         let validators = req.get_validators();
         debug!(self.log, "RPC request"; "endpoint" => "GetValidatorDuties", "epoch" => req.get_epoch());
 
-        let epoch = req.get_epoch();
+        let epoch = Epoch::from(req.get_epoch());
         let mut resp = GetDutiesResponse::new();
         let resp_validators = resp.mut_active_validators();
 
         let spec = self.chain.get_spec();
         let state = self.chain.get_state();
 
-        //TODO: Decide whether to rebuild the cache
+        let relative_epoch =
-        //TODO: Get the active validator indicies
+            match RelativeEpoch::from_epoch(state.slot.epoch(spec.slots_per_epoch), epoch) {
-        //let active_validator_indices = self.chain.state.read().get_cached_active_validator_indices(
+                Ok(v) => v,
-        let active_validator_indices = vec![1, 2, 3, 4, 5, 6, 7, 8];
+                Err(e) => {
-        // TODO: Is this the most efficient? Perhaps we cache this data structure.
+                    // incorrect epoch
+                    let log_clone = self.log.clone();
+                    let f = sink
+                        .fail(RpcStatus::new(
+                            RpcStatusCode::FailedPrecondition,
+                            Some(format!("Invalid epoch: {:?}", e)),
+                        ))
+                        .map_err(move |e| warn!(log_clone, "failed to reply {:?}: {:?}", req, e));
+                    return ctx.spawn(f);
+                }
+            };
 
-        // this is an array of validators who are to propose this epoch
+        let validator_proposers: Result<Vec<usize>, _> = epoch
-        // TODO: RelativeEpoch?
+            .slot_iter(spec.slots_per_epoch)
-        //let validator_proposers = [0..spec.slots_per_epoch].iter().map(|slot| state.get_beacon_proposer_index(Slot::from(slot), epoch, &spec)).collect();
+            .map(|slot| state.get_beacon_proposer_index(slot, relative_epoch, &spec))
-        let validator_proposers: Vec<u64> = vec![1, 2, 3, 4, 5];
+            .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() {
@@ -53,45 +78,65 @@ impl ValidatorService for ValidatorServiceInstance {
             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()),
                         ))
-                        //TODO: Handle error correctly
+                        .map_err(move |e| warn!(log_clone, "failed to reply {:?}", req));
-                        .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
                     return ctx.spawn(f);
                 }
             };
 
-            // is the validator active
+            // get the validator index
             let val_index = match state.get_validator_index(&public_key) {
-                Ok(Some(index)) => {
+                Ok(Some(index)) => index,
-                    if active_validator_indices.contains(&index) {
+                Ok(None) => {
-                        // validator is active, return the index
+                    // index not present in registry, set the duties for this key to None
-                        index
+                    warn!(
-                    } else {
+                        self.log,
-                        // validator is inactive, go to the next validator
+                        "RPC requested a public key that is not in the registry: {:?}", public_key
-                        active_validator.set_none(false);
+                    );
-                        resp_validators.push(active_validator);
-                        break;
-                    }
-                }
-                // validator index is not known, skip it
-                Ok(_) => {
                     active_validator.set_none(false);
                     resp_validators.push(active_validator);
-                    break;
+                    continue;
                 }
                 // the cache is not built, throw an error
-                Err(_) => {
+                Err(e) => {
+                    let log_clone = self.log.clone();
                     let f = sink
                         .fail(RpcStatus::new(
                             RpcStatusCode::FailedPrecondition,
-                            Some("Beacon state cache is not built".to_string()),
+                            Some(format!("Beacon state error {:?}", e)),
                         ))
-                        //TODO: Handle error correctly
+                        .map_err(move |e| warn!(log_clone, "Failed to reply {:?}: {:?}", req, e));
-                        .map_err(move |e| println!("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);
                 }
             };
@@ -100,33 +145,15 @@ impl ValidatorService for ValidatorServiceInstance {
             let mut duty = ValidatorDuty::new();
 
             // check if the validator needs to propose a block
-            if let Some(slot) = validator_proposers
+            if let Some(slot) = validator_proposers.iter().position(|&v| val_index == v) {
-                .iter()
+                duty.set_block_production_slot(
-                .position(|&v| val_index as u64 == v)
+                    epoch.start_slot(spec.slots_per_epoch).as_u64() + slot as u64,
-            {
+                );
-                duty.set_block_production_slot(epoch * spec.slots_per_epoch + slot as u64);
             } else {
                 // no blocks to propose this epoch
                 duty.set_none(false)
             }
 
-            // get attestation duties
-            let attestation_duties = match state.get_attestation_duties(val_index, &spec) {
-                Ok(Some(v)) => v,
-                Ok(_) => unreachable!(), //we've checked the validator index
-                // the cache is not built, throw an error
-                Err(_) => {
-                    let f = sink
-                        .fail(RpcStatus::new(
-                            RpcStatusCode::FailedPrecondition,
-                            Some("Beacon state cache is not built".to_string()),
-                        ))
-                        //TODO: Handle error correctly
-                        .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e));
-                    return ctx.spawn(f);
-                }
-            };
-
             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);

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

@@ -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(free_attestation)?;
         Ok(PollOutcome::AttestationProduced(slot))
     }
 

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/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

@@ -14,7 +14,7 @@ package ethereum.beacon.rpc.v1;
 
 // Service that currently identifies a beacon node
 service BeaconNodeService {
-    rpc Info(Empty) returns (NodeInfo);
+    rpc Info(Empty) returns (NodeInfoResponse);
 }
 
 /// Service that handles block production
@@ -40,11 +40,12 @@ service AttestationService {
 /*
  * Beacon Node Service Message
  */
-message NodeInfo {
+message NodeInfoResponse {
     string version = 1;
     Fork fork = 2;
     uint32 chain_id = 3;
     uint64 genesis_time = 4;
+    uint64 genesis_slot = 5;
 }
 
 message Fork {
@@ -53,8 +54,7 @@ message Fork {
     uint64 epoch = 3;
 }
 
-message Empty {
+message Empty {}
-}
 
 
 /*
@@ -83,10 +83,7 @@ message PublishBeaconBlockResponse {
 }
 
 message BeaconBlock {
-	uint64 slot = 1;
+	bytes ssz = 1;
-	bytes block_root = 2;
-	bytes randao_reveal = 3;
-	bytes signature = 4;
 }
 
 /*

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

@@ -1,6 +1,6 @@
-use clap::ArgMatches;
 use bincode;
 use bls::Keypair;
+use clap::ArgMatches;
 use slog::{debug, error, info};
 use std::fs;
 use std::fs::File;
@@ -67,6 +67,7 @@ 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!(),
             };

validator_client/src/duties/grpc.rs

@@ -1,9 +1,11 @@
 use super::epoch_duties::{EpochDuties, EpochDuty};
 use super::traits::{BeaconNode, BeaconNodeError};
+use grpcio::CallOption;
 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 {
@@ -21,6 +23,9 @@ impl BeaconNode for ValidatorServiceClient {
         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
             .get_validator_duties(&req)
@@ -31,7 +36,7 @@ impl BeaconNode for ValidatorServiceClient {
             if !validator_duty.has_duty() {
                 // validator is inactive
                 epoch_duties.insert(pubkeys[index].clone(), None);
-                break;
+                continue;
             }
             // active validator
             let active_duty = validator_duty.get_duty();

validator_client/src/duties/mod.rs

@@ -51,20 +51,23 @@ impl<U: BeaconNode> DutiesManager<U> {
     /// be a wall-clock (e.g., system time, remote server time, etc.).
     fn update(&self, epoch: Epoch) -> Result<UpdateOutcome, Error> {
         let duties = self.beacon_node.request_duties(epoch, &self.pubkeys)?;
-        // If these duties were known, check to see if they're updates or identical.
+        {
-        if let Some(known_duties) = self.duties_map.read()?.get(&epoch) {
+            // If these duties were known, check to see if they're updates or identical.
-            if *known_duties == duties {
+            if let Some(known_duties) = self.duties_map.read()?.get(&epoch) {
-                return Ok(UpdateOutcome::NoChange(epoch));
+                if *known_duties == duties {
-            } else {
+                    return Ok(UpdateOutcome::NoChange(epoch));
-                //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));
             }
-        } else {
+        }
+        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

validator_client/src/main.rs

@@ -43,8 +43,8 @@ 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();
 

validator_client/src/service.rs

@@ -43,8 +43,6 @@ pub struct Service {
     slot_clock: SystemTimeSlotClock,
     /// The current slot we are processing.
     current_slot: Slot,
-    /// Duration until the next slot. This is used for initializing the tokio timer interval.
-    duration_to_next_slot: Duration,
     /// The number of slots per epoch to allow for converting slots to epochs.
     slots_per_epoch: u64,
     // GRPC Clients
@@ -105,6 +103,7 @@ impl Service {
 
         // 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);
 
@@ -140,46 +139,21 @@ impl Service {
         };
 
         // build the validator slot clock
-        let slot_clock = SystemTimeSlotClock::new(genesis_time, config.spec.seconds_per_slot)
+        let slot_clock =
-            .expect("Unable to instantiate SystemTimeSlotClock.");
+            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");
 
-        // calculate the duration to the next slot
-        let duration_to_next_slot = {
-            let seconds_per_slot = config.spec.seconds_per_slot;
-            let syslot_time = SystemTime::now();
-            let duration_since_epoch = syslot_time
-                .duration_since(SystemTime::UNIX_EPOCH)
-                .map_err(|e| ErrorKind::SystemTimeError(e.to_string()))?;
-            let duration_since_genesis = duration_since_epoch
-                .checked_sub(Duration::from_secs(genesis_time))
-                .expect("Genesis must be in the future. Checked on connection");
-            let elapsed_slots = duration_since_epoch
-                .as_secs()
-                .checked_div(seconds_per_slot as u64)
-                .expect("Seconds per slot should not be 0");
-
-            // the duration to the next slot
-            Duration::from_secs(
-                (elapsed_slots + 1)
-                    .checked_mul(seconds_per_slot)
-                    .expect("Next slot time should not overflow u64"),
-            )
-            .checked_sub(duration_since_genesis)
-            .expect("This should never saturate")
-        };
-
         Ok(Self {
             connected_node_version: node_info.version,
             chain_id: node_info.chain_id as u16,
             fork,
             slot_clock,
             current_slot,
-            duration_to_next_slot,
             slots_per_epoch: config.spec.slots_per_epoch,
             beacon_block_client,
             validator_client,
@@ -202,15 +176,18 @@ impl Service {
             .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(
+            Interval::new(Instant::now() + duration_to_next_slot, slot_duration)
-                Instant::now() + service.duration_to_next_slot,
-                slot_duration,
-            )
         };
 
         /* kick off core service */
@@ -219,7 +196,7 @@ impl Service {
 
         // TODO: keypairs are randomly generated; they should be loaded from a file or generated.
         // https://github.com/sigp/lighthouse/issues/160
-        let keypairs = generate_deterministic_keypairs(8);
+        let keypairs = Arc::new(generate_deterministic_keypairs(8));
 
         /* build requisite objects to pass to core thread */
 
@@ -237,54 +214,59 @@ impl Service {
         });
 
         // run the core thread
-        runtime
+        runtime.block_on(
-            .block_on(interval.for_each(move |_| {
+            interval
-                let log = service.log.clone();
+                .for_each(move |_| {
+                    let log = service.log.clone();
 
-                /* get the current slot and epoch */
+                    /* get the current slot and epoch */
-                let current_slot = match service.slot_clock.present_slot() {
+                    let current_slot = match service.slot_clock.present_slot() {
-                    Err(e) => {
+                        Err(e) => {
-                        error!(log, "SystemTimeError {:?}", e);
+                            error!(log, "SystemTimeError {:?}", e);
-                        return Ok(());
+                            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();
-                tokio::spawn(futures::future::poll_fn(move || {
-                    cloned_manager.run_update(current_epoch.clone(), log.clone())
-                }));
-
-                /* 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() {
+                        Ok(slot) => slot.expect("Genesis is in the future"),
-                            // available AttestationDuty info
+                    };
-                            let attestation_duty =
+
-                                work_type.attestation_duty.expect("Cannot be None");
+                    let current_epoch = current_slot.epoch(service.slots_per_epoch);
-                            //TODO: Produce an attestation in a new thread
+
+                    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(())
+                    Ok(())
-            }))
+                })
-            .map_err(|e| format!("Service thread failed: {:?}", e))?;
+                .map_err(|e| format!("Service thread failed: {:?}", e)),
+        );
 
         // completed a slot process
         Ok(())
@@ -292,35 +274,6 @@ impl Service {
 
     /*
 
-    for keypair in keypairs {
-        info!(self.log, "Starting validator services"; "validator" => keypair.pk.concatenated_hex_id());
-
-        // 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 = self.slot_clock.clone();
-            let log = self.log.clone();
-            let beacon_node = self.validator_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();