From 1584469b7c754da9c45f36bf944ea14a21292da7 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Tue, 26 Mar 2019 17:41:43 +1100 Subject: [PATCH 1/8] Renamed attestation -> attestation_data for fetch, but not publish, to acknowledge the difference in the spec. Also started implementing the gRPC get_attestation_data functionality in the BeaconNode. --- .../validator_harness/direct_beacon_node.rs | 2 +- beacon_node/rpc/src/beacon_attester.rs | 59 +++++++++++++------ eth2/attester/src/lib.rs | 2 +- .../src/test_utils/simulated_beacon_node.rs | 2 +- eth2/attester/src/traits.rs | 2 +- protos/src/services.proto | 12 ++-- .../attestation_grpc_client.rs | 6 +- 7 files changed, 53 insertions(+), 32 deletions(-) diff --git a/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs b/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs index fde8211ab..17630833b 100644 --- a/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs +++ b/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs @@ -50,7 +50,7 @@ impl DirectBeaconNode { } impl AttesterBeaconNode for DirectBeaconNode { - fn produce_attestation( + fn produce_attestation_data( &self, _slot: Slot, shard: u64, diff --git a/beacon_node/rpc/src/beacon_attester.rs b/beacon_node/rpc/src/beacon_attester.rs index 36b6a40b2..88caacdd0 100644 --- a/beacon_node/rpc/src/beacon_attester.rs +++ b/beacon_node/rpc/src/beacon_attester.rs @@ -1,45 +1,66 @@ +use crate::beacon_chain::BeaconChain; use futures::Future; -use grpcio::{RpcContext, UnarySink}; +use grpcio::{RpcContext, UnarySink, RpcStatus, RpcStatusCode}; use protos::services::{ - Attestation as AttestationProto, ProduceAttestation, ProduceAttestationResponse, - ProduceAttestationRequest, PublishAttestationResponse, PublishAttestationRequest, + AttestationData as AttestationDataProto, ProduceAttestationData, ProduceAttestationDataResponse, + ProduceAttestationDataRequest, PublishAttestationResponse, PublishAttestationRequest, PublishAttestation }; use protos::services_grpc::BeaconBlockService; -use slog::Logger; +use slog::{Logger, info, warn, error}; #[derive(Clone)] pub struct AttestationServiceInstance { + pub chain: Arc, pub log: Logger, } impl AttestationService for AttestationServiceInstance { - /// Produce a `BeaconBlock` for signing by a validator. - fn produce_attestation( + /// Produce the `AttestationData` for signing by a validator. + fn produce_attestation_data( &mut self, ctx: RpcContext, - req: ProduceAttestationRequest, - sink: UnarySink, + req: ProduceAttestationDataRequest, + sink: UnarySink, ) { - println!("producing attestation at slot {}", req.get_slot()); + info!(&self.log, "Attempting to produce 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()); + // get the chain spec & state + let spec = self.chain.get_spec(); + let state = self.chain.get_state(); - let mut resp = ProduceAttestationResponse::new(); - resp.set_attestation_data(attestation); + // Start by performing some checks + // Check that the the AttestionData is for the current slot (otherwise it will not be valid) + if req.get_slot() != state.slot { + let f = sink + .fail(RpcStatus::new( + RpcStatusCode::OutOfRange, + "AttestationData request for a slot that is not the current slot." + )) + .map_err(move |e| error!(&self.log, "Failed to reply with failure {:?}: {:?}", req, e)); + } + + // Then collect the data we need for the AttesatationData object + //let beacon_block_root = state.latest_block_roots.first().ok_or_else(|e| ) + + // And finally build the AttestationData object + let mut attestation_data = AttestationDataProto::new(); + attestation_data.set_slot(state.slot.as_u64()); + attestation_data.set_shard(spec.genesis_start_shard); + attestation_data.set_beacon_block_root(b"cats".to_vec()); + //attestation_data. + + let mut resp = ProduceAttestationDataResponse::new(); + resp.set_attestation_data(attestation_data); let f = sink .success(resp) - .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e)); + .map_err(move |e| error!("Failed to reply with success {:?}: {:?}", req, e)); ctx.spawn(f) } - /// Accept some fully-formed `BeaconBlock`, process and publish it. + /// Accept some fully-formed `FreeAttestation` from the validator, + /// store it, and aggregate it into an `Attestation`. fn publish_attestation( &mut self, ctx: RpcContext, diff --git a/eth2/attester/src/lib.rs b/eth2/attester/src/lib.rs index 065fdc923..3f13555e3 100644 --- a/eth2/attester/src/lib.rs +++ b/eth2/attester/src/lib.rs @@ -94,7 +94,7 @@ impl Attester Result { - let attestation_data = match self.beacon_node.produce_attestation(slot, shard)? { + let attestation_data = match self.beacon_node.produce_attestation_data(slot, shard)? { Some(attestation_data) => attestation_data, None => return Ok(PollOutcome::BeaconNodeUnableToProduceAttestation(slot)), }; diff --git a/eth2/attester/src/test_utils/simulated_beacon_node.rs b/eth2/attester/src/test_utils/simulated_beacon_node.rs index 84a203cdb..d19f43422 100644 --- a/eth2/attester/src/test_utils/simulated_beacon_node.rs +++ b/eth2/attester/src/test_utils/simulated_beacon_node.rs @@ -26,7 +26,7 @@ impl SimulatedBeaconNode { } impl BeaconNode for SimulatedBeaconNode { - fn produce_attestation(&self, slot: Slot, shard: u64) -> ProduceResult { + fn produce_attestation_data(&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(), diff --git a/eth2/attester/src/traits.rs b/eth2/attester/src/traits.rs index 749c6e1a2..2fd6940af 100644 --- a/eth2/attester/src/traits.rs +++ b/eth2/attester/src/traits.rs @@ -14,7 +14,7 @@ pub enum PublishOutcome { /// Defines the methods required to produce and publish blocks on a Beacon Node. pub trait BeaconNode: Send + Sync { - fn produce_attestation( + fn produce_attestation_data( &self, slot: Slot, shard: u64, diff --git a/protos/src/services.proto b/protos/src/services.proto index 80d512c54..35b9c32af 100644 --- a/protos/src/services.proto +++ b/protos/src/services.proto @@ -33,7 +33,7 @@ service ValidatorService { /// Service that handles validator attestations service AttestationService { - rpc ProduceAttestation(ProduceAttestationRequest) returns (ProduceAttestationResponse); + rpc ProduceAttestation(ProduceAttestationDataRequest) returns (ProduceAttestationDataResponse); rpc PublishAttestation(PublishAttestationRequest) returns (PublishAttestationResponse); } @@ -138,13 +138,13 @@ message ProposeBlockSlotResponse { * Attestation Service Messages */ -message ProduceAttestationRequest { +message ProduceAttestationDataRequest { uint64 slot = 1; uint64 shard = 2; } -message ProduceAttestationResponse { - Attestation attestation_data = 1; +message ProduceAttestationDataResponse { + AttestationData attestation_data = 1; } message PublishAttestationRequest { @@ -162,7 +162,7 @@ message Crosslink { } -message Attestation { +message AttestationData { uint64 slot = 1; uint64 shard = 2; bytes beacon_block_root = 3; @@ -175,7 +175,7 @@ message Attestation { } message FreeAttestation { - Attestation attestation_data = 1; + AttestationData data = 1; bytes signature = 2; uint64 validator_index = 3; } diff --git a/validator_client/src/attester_service/attestation_grpc_client.rs b/validator_client/src/attester_service/attestation_grpc_client.rs index 5a4701ba9..f55acc4f7 100644 --- a/validator_client/src/attester_service/attestation_grpc_client.rs +++ b/validator_client/src/attester_service/attestation_grpc_client.rs @@ -2,7 +2,7 @@ use protos::services_grpc::AttestationServiceClient; use std::sync::Arc; use attester::{BeaconNode, BeaconNodeError, PublishOutcome}; -use protos::services::ProduceAttestationRequest; +use protos::services::ProduceAttestationDataRequest; use types::{AttestationData, FreeAttestation, Slot}; pub struct AttestationGrpcClient { @@ -16,12 +16,12 @@ impl AttestationGrpcClient { } impl BeaconNode for AttestationGrpcClient { - fn produce_attestation( + fn produce_attestation_data( &self, slot: Slot, shard: u64, ) -> Result, BeaconNodeError> { - let mut req = ProduceAttestationRequest::new(); + let mut req = ProduceAttestationDataRequest::new(); req.set_slot(slot.as_u64()); req.set_shard(shard); From c9a7977d6998183458ab7a5d365ac08d8fa69257 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Wed, 27 Mar 2019 14:30:09 +1100 Subject: [PATCH 2/8] Renamed some functions, trying to get beaconnode attestation stuff to work. --- beacon_node/beacon_chain/src/beacon_chain.rs | 2 +- beacon_node/rpc/src/beacon_attester.rs | 17 +++++++---------- protos/src/services.proto | 2 +- .../attester_service/attestation_grpc_client.rs | 2 +- 4 files changed, 10 insertions(+), 13 deletions(-) diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 61b5fb58b..110d1a99d 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -280,7 +280,7 @@ where } /// Produce an `AttestationData` that is valid for the present `slot` and given `shard`. - pub fn produce_attestation(&self, shard: u64) -> Result { + pub fn produce_attestation_data(&self, shard: u64) -> Result { trace!("BeaconChain::produce_attestation: shard: {}", shard); let source_epoch = self.state.read().current_justified_epoch; let source_root = *self.state.read().get_block_root( diff --git a/beacon_node/rpc/src/beacon_attester.rs b/beacon_node/rpc/src/beacon_attester.rs index 88caacdd0..4166c30d4 100644 --- a/beacon_node/rpc/src/beacon_attester.rs +++ b/beacon_node/rpc/src/beacon_attester.rs @@ -9,6 +9,8 @@ use protos::services::{ use protos::services_grpc::BeaconBlockService; use slog::{Logger, info, warn, error}; +const TEST_SHARD_PHASE_ZERO: u8 = 0; + #[derive(Clone)] pub struct AttestationServiceInstance { pub chain: Arc, @@ -29,9 +31,11 @@ impl AttestationService for AttestationServiceInstance { let spec = self.chain.get_spec(); let state = self.chain.get_state(); + let slot_requested = req.get_slot(); + // Start by performing some checks // Check that the the AttestionData is for the current slot (otherwise it will not be valid) - if req.get_slot() != state.slot { + if slot_requested != state.slot { let f = sink .fail(RpcStatus::new( RpcStatusCode::OutOfRange, @@ -40,15 +44,8 @@ impl AttestationService for AttestationServiceInstance { .map_err(move |e| error!(&self.log, "Failed to reply with failure {:?}: {:?}", req, e)); } - // Then collect the data we need for the AttesatationData object - //let beacon_block_root = state.latest_block_roots.first().ok_or_else(|e| ) - - // And finally build the AttestationData object - let mut attestation_data = AttestationDataProto::new(); - attestation_data.set_slot(state.slot.as_u64()); - attestation_data.set_shard(spec.genesis_start_shard); - attestation_data.set_beacon_block_root(b"cats".to_vec()); - //attestation_data. + // Then get the AttestationData from the beacon chain (for shard 0 for now) + let attestation_data = self.chain.produce_attestation_data(TEST_SHARD_PHASE_ZERO); let mut resp = ProduceAttestationDataResponse::new(); resp.set_attestation_data(attestation_data); diff --git a/protos/src/services.proto b/protos/src/services.proto index 35b9c32af..1dfb53c06 100644 --- a/protos/src/services.proto +++ b/protos/src/services.proto @@ -33,7 +33,7 @@ service ValidatorService { /// Service that handles validator attestations service AttestationService { - rpc ProduceAttestation(ProduceAttestationDataRequest) returns (ProduceAttestationDataResponse); + rpc ProduceAttestationData(ProduceAttestationDataRequest) returns (ProduceAttestationDataResponse); rpc PublishAttestation(PublishAttestationRequest) returns (PublishAttestationResponse); } diff --git a/validator_client/src/attester_service/attestation_grpc_client.rs b/validator_client/src/attester_service/attestation_grpc_client.rs index f55acc4f7..bfb7f67c6 100644 --- a/validator_client/src/attester_service/attestation_grpc_client.rs +++ b/validator_client/src/attester_service/attestation_grpc_client.rs @@ -27,7 +27,7 @@ impl BeaconNode for AttestationGrpcClient { let reply = self .client - .produce_attestation(&req) + .produce_attestation_data(&req) .map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?; // TODO: return correct Attestation From bda381a2648fa21d420b14703bffa55539fc44d7 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Thu, 28 Mar 2019 09:38:39 +1100 Subject: [PATCH 3/8] More progress towards getting the attester working. --- eth2/attester/src/lib.rs | 2 + validator_client/src/service.rs | 164 +++++++++++++++++--------------- 2 files changed, 87 insertions(+), 79 deletions(-) diff --git a/eth2/attester/src/lib.rs b/eth2/attester/src/lib.rs index 3f13555e3..7b1d26145 100644 --- a/eth2/attester/src/lib.rs +++ b/eth2/attester/src/lib.rs @@ -99,6 +99,8 @@ impl Attester return Ok(PollOutcome::BeaconNodeUnableToProduceAttestation(slot)), }; + dbg!(&attestation_data); + if !self.safe_to_produce(&attestation_data) { return Ok(PollOutcome::SlashableAttestationNotProduced(slot)); } diff --git a/validator_client/src/service.rs b/validator_client/src/service.rs index 8a7e90d10..4c6a49997 100644 --- a/validator_client/src/service.rs +++ b/validator_client/src/service.rs @@ -25,9 +25,12 @@ use tokio::runtime::Builder; use tokio::timer::Interval; use tokio_timer::clock::Clock; use types::{Epoch, Fork, Slot}; +use std::thread; //TODO: This service should be simplified in the future. Can be made more steamlined. +const POLL_INTERVAL_MILLIS: u64 = 100; + /// The validator service. This is the main thread that executes and maintains validator /// duties. pub struct Service { @@ -217,7 +220,10 @@ 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 = Arc::new(vec![Keypair::random()]); + let keypairs = match config.fetch_keys(&log) { + Some(kps) => kps, + None => panic!("No key pairs found, cannot start validator client without. Try running ./account_manager generate first.") + }; // build requisite objects to pass to core thread. let duties_map = Arc::new(EpochDutiesMap::new(config.spec.slots_per_epoch)); @@ -272,87 +278,87 @@ impl Service { Ok(()) })) .map_err(|e| format!("Service thread failed: {:?}", e))?; + + let mut threads = vec![]; + + for keypair in keypairs { + info!(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(); + 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 slot_clock = service.slot_clock.clone(); + let log = log.clone(); + let attester_grpc_client = Arc::new(AttestationGrpcClient::new(attester_client.clone())); + thread::spawn(move || { + let attester = Attester::new(epoch_map_for_attester, slot_clock, attester_grpc_client, signer); + let mut attester_service = AttesterService { + attester, + poll_interval_millis, + log, + }; + + attester_service.run(); + }) + }; + + //threads.push((duties_manager_thread, producer_thread, attester_thread)); + threads.push((attester_thread)); + } + Ok(()) - } + } /* - - let duties_map = Arc::new(EpochDutiesMap::new(spec.slots_per_epoch)); - let epoch_map_for_attester = Arc::new(EpochMap::new(spec.slots_per_epoch)); - - - 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(); - 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; From ba71e8adca2a6f1e899492bdcd776dc45f981cc3 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Thu, 28 Mar 2019 20:55:07 +1100 Subject: [PATCH 4/8] Merged age-validator-client into luke's changes on validator_client, and fixed all the merge conflicts. --- Jenkinsfile | 22 ++- beacon_node/Cargo.toml | 1 + beacon_node/beacon_chain/src/beacon_chain.rs | 23 +++ beacon_node/beacon_chain/src/initialise.rs | 20 +- beacon_node/client/Cargo.toml | 1 + beacon_node/client/src/lib.rs | 101 +++++++++- beacon_node/client/src/notifier.rs | 4 +- beacon_node/eth2-libp2p/src/behaviour.rs | 33 +++- beacon_node/network/src/sync/simple_sync.rs | 8 +- beacon_node/rpc/src/beacon_block.rs | 110 ++++++++--- beacon_node/rpc/src/beacon_chain.rs | 13 +- beacon_node/rpc/src/beacon_node.rs | 7 +- beacon_node/rpc/src/lib.rs | 1 + beacon_node/rpc/src/validator.rs | 135 ++++++++------ beacon_node/src/run.rs | 2 + eth2/attester/src/lib.rs | 3 +- .../src/per_slot_processing.rs | 1 - .../testing_beacon_state_builder.rs | 4 +- eth2/utils/slot_clock/src/lib.rs | 3 + .../slot_clock/src/system_time_slot_clock.rs | 39 +++- .../slot_clock/src/testing_slot_clock.rs | 6 + protos/src/services.proto | 13 +- .../beacon_block_grpc_client.rs | 38 +--- validator_client/src/config.rs | 3 +- validator_client/src/duties/grpc.rs | 7 +- validator_client/src/duties/mod.rs | 23 ++- validator_client/src/main.rs | 4 +- validator_client/src/service.rs | 173 ++++++++---------- 28 files changed, 540 insertions(+), 258 deletions(-) diff --git a/Jenkinsfile b/Jenkinsfile index 42755d5f7..1a3afad87 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -1,20 +1,22 @@ pipeline { - agent { + agent { dockerfile { filename 'Dockerfile' args '-v cargo-cache:/cargocache:rw -e "CARGO_HOME=/cargocache"' } } - stages { - stage('Build') { - steps { - sh 'cargo build' - } - } - stage('Test') { + stages { + stage('Build') { 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' + } + } + } } diff --git a/beacon_node/Cargo.toml b/beacon_node/Cargo.toml index e7aaf938d..a090c1cc5 100644 --- a/beacon_node/Cargo.toml +++ b/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" } diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 9323d1334..a9a28ea39 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/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 { + 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 diff --git a/beacon_node/beacon_chain/src/initialise.rs b/beacon_node/beacon_chain/src/initialise.rs index 7d3c87965..0951e06fb 100644 --- a/beacon_node/beacon_chain/src/initialise.rs +++ b/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) - .expect("Unable to load SystemTimeSlotClock"); + let slot_clock = SystemTimeSlotClock::new( + 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) - .expect("Unable to load SystemTimeSlotClock"); + let slot_clock = SystemTimeSlotClock::new( + 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()); diff --git a/beacon_node/client/Cargo.toml b/beacon_node/client/Cargo.toml index 12c1b5c80..8956dbb07 100644 --- a/beacon_node/client/Cargo.toml +++ b/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" diff --git a/beacon_node/client/src/lib.rs b/beacon_node/client/src/lib.rs index b24d2cb7f..807fd9301 100644 --- a/beacon_node/client/src/lib.rs +++ b/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 { pub network: Arc, /// Signal to terminate the RPC server. pub rpc_exit_signal: Option, + /// Signal to terminate the slot timer. + pub slot_timer_exit_signal: Option, /// The clients logger. log: slog::Logger, /// Marker to pin the beacon chain generics. @@ -42,6 +52,35 @@ impl Client { // 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 Client { )); } + 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(chain: &Arc>, 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 + ), + }; + } +} diff --git a/beacon_node/client/src/notifier.rs b/beacon_node/client/src/notifier.rs index 335183c7d..91a9f3a26 100644 --- a/beacon_node/client/src/notifier.rs +++ b/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(client: &Client, 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; diff --git a/beacon_node/eth2-libp2p/src/behaviour.rs b/beacon_node/eth2-libp2p/src/behaviour.rs index 41b7c8965..286597183 100644 --- a/beacon_node/eth2-libp2p/src/behaviour.rs +++ b/beacon_node/eth2-libp2p/src/behaviour.rs @@ -49,12 +49,15 @@ impl NetworkBehaviourEventProcess { + 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); + } +} diff --git a/beacon_node/network/src/sync/simple_sync.rs b/beacon_node/network/src/sync/simple_sync.rs index 2aa0a1d7d..85949fa98 100644 --- a/beacon_node/network/src/sync/simple_sync.rs +++ b/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), ); diff --git a/beacon_node/rpc/src/beacon_block.rs b/beacon_node/rpc/src/beacon_block.rs index 4e1875665..f6b426c18 100644 --- a/beacon_node/rpc/src/beacon_block.rs +++ b/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, pub network_chan: crossbeam_channel::Sender, 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_block_root(b"cats".to_vec()); + block.set_ssz(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, ) { - 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) diff --git a/beacon_node/rpc/src/beacon_chain.rs b/beacon_node/rpc/src/beacon_chain.rs index 9b2681876..0551a8024 100644 --- a/beacon_node/rpc/src/beacon_chain.rs +++ b/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; + + fn process_block(&self, block: BeaconBlock) + -> Result; } impl BeaconChain for RawBeaconChain @@ -28,4 +32,11 @@ where fn get_state(&self) -> RwLockReadGuard { self.state.read() } + + fn process_block( + &self, + block: BeaconBlock, + ) -> Result { + self.process_block(block) + } } diff --git a/beacon_node/rpc/src/beacon_node.rs b/beacon_node/rpc/src/beacon_node.rs index 7b00f04f4..c92ab6616 100644 --- a/beacon_node/rpc/src/beacon_node.rs +++ b/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) { + fn info(&mut self, ctx: RpcContext, _req: Empty, sink: UnarySink) { 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 diff --git a/beacon_node/rpc/src/lib.rs b/beacon_node/rpc/src/lib.rs index 20cd62b1d..2d47b4a69 100644 --- a/beacon_node/rpc/src/lib.rs +++ b/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(), }; diff --git a/beacon_node/rpc/src/validator.rs b/beacon_node/rpc/src/validator.rs index 47886a9df..936c95f52 100644 --- a/beacon_node/rpc/src/validator.rs +++ b/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 - //TODO: Get the active validator indicies - //let active_validator_indices = self.chain.state.read().get_cached_active_validator_indices( - let active_validator_indices = vec![1, 2, 3, 4, 5, 6, 7, 8]; - // TODO: Is this the most efficient? Perhaps we cache this data structure. + let relative_epoch = + match RelativeEpoch::from_epoch(state.slot.epoch(spec.slots_per_epoch), epoch) { + Ok(v) => v, + Err(e) => { + // 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 - // TODO: RelativeEpoch? - //let validator_proposers = [0..spec.slots_per_epoch].iter().map(|slot| state.get_beacon_proposer_index(Slot::from(slot), epoch, &spec)).collect(); - let validator_proposers: Vec = vec![1, 2, 3, 4, 5]; + let validator_proposers: Result, _> = 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() { @@ -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| println!("failed to reply {:?}: {:?}", req, e)); + .map_err(move |e| warn!(log_clone, "failed to reply {:?}", req)); 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)) => { - if active_validator_indices.contains(&index) { - // validator is active, return the index - index - } else { - // validator is inactive, go to the next validator - active_validator.set_none(false); - resp_validators.push(active_validator); - break; - } - } - // validator index is not known, skip it - Ok(_) => { + 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); - 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| println!("failed to reply {:?}: {:?}", req, 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); } }; @@ -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 - .iter() - .position(|&v| val_index as u64 == v) - { - duty.set_block_production_slot(epoch * spec.slots_per_epoch + slot as u64); + 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) } - // 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); diff --git a/beacon_node/src/run.rs b/beacon_node/src/run.rs index b3b284452..1d9156124 100644 --- a/beacon_node/src/run.rs +++ b/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))?; diff --git a/eth2/attester/src/lib.rs b/eth2/attester/src/lib.rs index 8dd83fa04..53d81e897 100644 --- a/eth2/attester/src/lib.rs +++ b/eth2/attester/src/lib.rs @@ -90,8 +90,7 @@ impl Attester { aggregate_signature: agg_sig, }; - self.beacon_node - .publish_attestation(attestation)?; + self.beacon_node.publish_attestation(attestation)?; Ok(PollOutcome::AttestationProduced(attestation_duty.slot)) } diff --git a/eth2/state_processing/src/per_slot_processing.rs b/eth2/state_processing/src/per_slot_processing.rs index 8f02b70e3..c6b5312c7 100644 --- a/eth2/state_processing/src/per_slot_processing.rs +++ b/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; diff --git a/eth2/types/src/test_utils/testing_beacon_state_builder.rs b/eth2/types/src/test_utils/testing_beacon_state_builder.rs index 473cd4166..b38e8b527 100644 --- a/eth2/types/src/test_utils/testing_beacon_state_builder.rs +++ b/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(), diff --git a/eth2/utils/slot_clock/src/lib.rs b/eth2/utils/slot_clock/src/lib.rs index 0379d50d9..fd5a2d1d7 100644 --- a/eth2/utils/slot_clock/src/lib.rs +++ b/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, Self::Error>; + + fn duration_to_next_slot(&self) -> Result, Self::Error>; } diff --git a/eth2/utils/slot_clock/src/system_time_slot_clock.rs b/eth2/utils/slot_clock/src/system_time_slot_clock.rs index 99f051985..4dfc6b37d 100644 --- a/eth2/utils/slot_clock/src/system_time_slot_clock.rs +++ b/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 { @@ -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, Error> { + duration_to_next_slot(self.genesis_seconds, self.slot_duration_seconds) + } } impl From 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, 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, }; diff --git a/eth2/utils/slot_clock/src/testing_slot_clock.rs b/eth2/utils/slot_clock/src/testing_slot_clock.rs index 80ee40539..b5c36dfa0 100644 --- a/eth2/utils/slot_clock/src/testing_slot_clock.rs +++ b/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, Error> { + Ok(Some(Duration::from_secs(1))) + } } #[cfg(test)] diff --git a/protos/src/services.proto b/protos/src/services.proto index bdb311fd6..b61e5aac0 100644 --- a/protos/src/services.proto +++ b/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 block_root = 2; - bytes randao_reveal = 3; - bytes signature = 4; + bytes ssz = 1; } /* diff --git a/validator_client/src/block_producer_service/beacon_block_grpc_client.rs b/validator_client/src/block_producer_service/beacon_block_grpc_client.rs index 04a02a221..ba2acfffb 100644 --- a/validator_client/src/block_producer_service/beacon_block_grpc_client.rs +++ b/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) - .map_err(|_| BeaconNodeError::DecodeFailure)?; + let (block, _i) = + BeaconBlock::ssz_decode(&ssz, 0).map_err(|_| BeaconNodeError::DecodeFailure)?; - let (randao_reveal, _) = Signature::ssz_decode(block.get_randao_reveal(), 0) - .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![], - }, - })) + Ok(Some(block)) } else { Ok(None) } @@ -79,12 +58,11 @@ impl BeaconNode for BeaconBlockGrpcClient { fn publish_beacon_block(&self, block: BeaconBlock) -> Result { 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_block_root(vec![0]); - grpc_block.set_randao_reveal(ssz_encode(&block.body.randao_reveal)); - grpc_block.set_signature(ssz_encode(&block.signature)); + grpc_block.set_ssz(ssz); req.set_block(grpc_block); diff --git a/validator_client/src/config.rs b/validator_client/src/config.rs index 78900374a..3d426e8c7 100644 --- a/validator_client/src/config.rs +++ b/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!(), }; diff --git a/validator_client/src/duties/grpc.rs b/validator_client/src/duties/grpc.rs index 4bb76e14c..041d41add 100644 --- a/validator_client/src/duties/grpc.rs +++ b/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, Keypair, Slot}; impl BeaconNode for ValidatorServiceClient { @@ -21,6 +23,9 @@ impl BeaconNode for ValidatorServiceClient { validators.set_public_keys(signers.iter().map(|v| ssz_encode(&v.pk)).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(signers[index].clone(), None); - break; + continue; } // active validator let active_duty = validator_duty.get_duty(); diff --git a/validator_client/src/duties/mod.rs b/validator_client/src/duties/mod.rs index e239ca795..9d9221592 100644 --- a/validator_client/src/duties/mod.rs +++ b/validator_client/src/duties/mod.rs @@ -52,20 +52,23 @@ impl DutiesManager { /// be a wall-clock (e.g., system time, remote server time, etc.). fn update(&self, epoch: Epoch) -> Result { let duties = self.beacon_node.request_duties(epoch, &self.signers)?; - // 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 *known_duties == duties { - return Ok(UpdateOutcome::NoChange(epoch)); - } else { - //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)); + { + // 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 *known_duties == duties { + return Ok(UpdateOutcome::NoChange(epoch)); + } } - } 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 diff --git a/validator_client/src/main.rs b/validator_client/src/main.rs index 84d0cbff7..d044030fe 100644 --- a/validator_client/src/main.rs +++ b/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"]) - .default_value("foundation"), + .possible_values(&["foundation", "few_validators", "lighthouse_testnet"]) + .default_value("lighthouse_testnet"), ) .get_matches(); diff --git a/validator_client/src/service.rs b/validator_client/src/service.rs index c2caed59d..86f310a06 100644 --- a/validator_client/src/service.rs +++ b/validator_client/src/service.rs @@ -25,6 +25,7 @@ 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}; use std::thread; @@ -45,8 +46,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 @@ -107,6 +106,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); @@ -142,46 +142,21 @@ impl Service { }; // build the validator slot clock - let slot_clock = SystemTimeSlotClock::new(genesis_time, config.spec.seconds_per_slot) - .expect("Unable to instantiate SystemTimeSlotClock."); + 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"); - // 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, @@ -204,15 +179,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( - Instant::now() + service.duration_to_next_slot, - slot_duration, - ) + Interval::new(Instant::now() + duration_to_next_slot, slot_duration) }; /* kick off core service */ @@ -221,11 +199,14 @@ impl Service { // TODO: keypairs are randomly generated; they should be loaded from a file or generated. // https://github.com/sigp/lighthouse/issues/160 + /* In future, load generated keys from disk. let keypairs = match config.fetch_keys(&log.clone()) { Some(kps) => kps, None => panic!("No key pairs found, cannot start validator client without. Try running ./account_manager generate first.") }; + */ + let keypairs = Arc::new(generate_deterministic_keypairs(8)); /* build requisite objects to pass to core thread */ @@ -243,71 +224,75 @@ impl Service { }); // run the core thread - runtime - .block_on(interval.for_each(move |_| { - let log = service.log.clone(); + 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_log = log.clone(); - let cloned_manager = manager.clone(); - tokio::spawn(futures::future::poll_fn(move || { - cloned_manager.run_update(current_epoch.clone(), cloned_log.clone()) - })); - - /* execute any specified duties */ - - if let Some(work) = manager.get_current_work(current_slot) { - for (keypair, work_type) in work { - if work_type.produce_block { - // TODO: Produce a beacon block in a new thread + /* get the current slot and epoch */ + let current_slot = match service.slot_clock.present_slot() { + Err(e) => { + error!(log, "SystemTimeError {:?}", e); + return Ok(()); } - if work_type.attestation_duty.is_some() { - // available AttestationDuty info - let attestation_duty = - work_type.attestation_duty.expect("Cannot be None"); - let attester_grpc_client = - Arc::new( - AttestationGrpcClient::new( - service.attester_client.clone() - ) - ); - let signer = Arc::new(AttesterLocalSigner::new(keypair.clone())); - let attester = - Attester::new( - attester_grpc_client, - signer); - let mut attester_service = AttesterService { - attester, - poll_interval_millis: POLL_INTERVAL_MILLIS, - log: log.clone(), - }; - attester_service.run(); + 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 (keypair, 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"); + let attester_grpc_client = + Arc::new( + AttestationGrpcClient::new( + service.attester_client.clone() + ) + ); + let signer = Arc::new(AttesterLocalSigner::new(keypair.clone())); + let attester = + Attester::new( + attester_grpc_client, + signer); + let mut attester_service = AttesterService { + attester, + poll_interval_millis: POLL_INTERVAL_MILLIS, + log: log.clone(), + }; + attester_service.run(); + } } } - } - Ok(()) - })) - .map_err(|e| format!("Service thread failed: {:?}", e))?; + Ok(()) + }) + .map_err(|e| format!("Service thread failed: {:?}", e)), + ); // completed a slot process Ok(()) From 867af4bc6ad5efcaeba3d6891031c2bcb3997a8c Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Thu, 28 Mar 2019 21:00:38 +1100 Subject: [PATCH 5/8] Made the 'signers' an Arc, so that things compile. --- validator_client/src/duties/mod.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/validator_client/src/duties/mod.rs b/validator_client/src/duties/mod.rs index 9d9221592..0b86e3ce9 100644 --- a/validator_client/src/duties/mod.rs +++ b/validator_client/src/duties/mod.rs @@ -42,7 +42,7 @@ pub struct DutiesManager { pub duties_map: RwLock, /// A list of all signer objects known to the validator service. // TODO: Generalise the signers, so that they're not just keypairs - pub signers: Vec, + pub signers: Arc>, pub beacon_node: Arc, } @@ -97,7 +97,7 @@ impl DutiesManager { // if the map is poisoned, return None let duties = self.duties_map.read().ok()?; - for validator_signer in &self.signers { + for validator_signer in self.signers.iter() { match duties.is_work_slot(slot, &validator_signer) { Ok(Some(work_type)) => current_work.push((validator_signer.clone(), work_type)), Ok(None) => {} // No work for this validator From 87acaac8a03b2735b20e64887c7d5e0a5627f55a Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Thu, 28 Mar 2019 21:01:47 +1100 Subject: [PATCH 6/8] Ran cargo fmt. --- .../test_harness/src/validator_harness/mod.rs | 8 ++-- eth2/attester/src/lib.rs | 37 +++++++++++-------- .../src/test_utils/simulated_beacon_node.rs | 2 +- eth2/attester/src/traits.rs | 2 +- eth2/block_proposer/src/lib.rs | 18 ++------- .../attestation_grpc_client.rs | 2 +- validator_client/src/duties/epoch_duties.rs | 2 +- validator_client/src/duties/mod.rs | 2 +- validator_client/src/service.rs | 16 +++----- 9 files changed, 38 insertions(+), 51 deletions(-) diff --git a/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs b/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs index 074ede06a..43ad03ea7 100644 --- a/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs +++ b/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs @@ -2,13 +2,13 @@ mod direct_beacon_node; mod direct_duties; mod local_signer; +use crate::direct_beacon_node::DirectBeaconNode; use attester::PollOutcome as AttestationPollOutcome; use attester::{Attester, Error as AttestationPollError}; use beacon_chain::BeaconChain; use block_proposer::PollOutcome as BlockPollOutcome; use block_proposer::{BlockProducer, Error as BlockPollError}; use db::MemoryDB; -use crate::direct_beacon_node::DirectBeaconNode; use fork_choice::BitwiseLMDGhost; use local_signer::LocalSigner; use slot_clock::TestingSlotClock; @@ -32,10 +32,8 @@ type TestingBlockProducer = BlockProducer< LocalSigner, >; -type TestingAttester = Attester< - DirectBeaconNode>, - LocalSigner, ->; +type TestingAttester = + Attester>, LocalSigner>; /// A `BlockProducer` and `Attester` which sign using a common keypair. /// diff --git a/eth2/attester/src/lib.rs b/eth2/attester/src/lib.rs index 53d81e897..b5cdf5920 100644 --- a/eth2/attester/src/lib.rs +++ b/eth2/attester/src/lib.rs @@ -3,8 +3,10 @@ mod traits; use ssz::TreeHash; use std::sync::Arc; -use types::{AttestationData, AttestationDataAndCustodyBit, Attestation, Signature, - AggregateSignature, Slot, AttestationDuty, Bitfield}; +use types::{ + AggregateSignature, Attestation, AttestationData, AttestationDataAndCustodyBit, + AttestationDuty, Bitfield, Signature, Slot, +}; pub use self::traits::{ BeaconNode, BeaconNodeError, DutiesReader, DutiesReaderError, PublishOutcome, Signer, @@ -59,20 +61,28 @@ impl Attester { } impl Attester { - - fn produce_attestation(&mut self, attestation_duty: AttestationDuty) -> Result { - let attestation_data = match self.beacon_node.produce_attestation_data( - attestation_duty.slot, - attestation_duty.shard - )? { + fn produce_attestation( + &mut self, + attestation_duty: AttestationDuty, + ) -> Result { + let attestation_data = match self + .beacon_node + .produce_attestation_data(attestation_duty.slot, attestation_duty.shard)? + { Some(attestation_data) => attestation_data, - None => return Ok(PollOutcome::BeaconNodeUnableToProduceAttestation(attestation_duty.slot)), + None => { + return Ok(PollOutcome::BeaconNodeUnableToProduceAttestation( + attestation_duty.slot, + )) + } }; dbg!(&attestation_data); if !self.safe_to_produce(&attestation_data) { - return Ok(PollOutcome::SlashableAttestationNotProduced(attestation_duty.slot)); + return Ok(PollOutcome::SlashableAttestationNotProduced( + attestation_duty.slot, + )); } let signature = match self.sign_attestation_data(&attestation_data) { @@ -82,7 +92,6 @@ impl Attester { let mut agg_sig = AggregateSignature::new(); agg_sig.add(&signature); - let attestation = Attestation { aggregation_bitfield: Bitfield::new(), data: attestation_data, @@ -172,10 +181,7 @@ mod tests { let attest_epoch = attest_slot / spec.slots_per_epoch; let attest_shard = 12; - let mut attester = Attester::new( - beacon_node.clone(), - signer.clone(), - ); + let mut attester = Attester::new(beacon_node.clone(), signer.clone()); // Configure responses from the BeaconNode. beacon_node.set_next_produce_result(Ok(Some(AttestationData::random_for_test(&mut rng)))); @@ -220,6 +226,5 @@ mod tests { Ok(PollOutcome::ProducerDutiesUnknown(slot)) ); */ - } } diff --git a/eth2/attester/src/test_utils/simulated_beacon_node.rs b/eth2/attester/src/test_utils/simulated_beacon_node.rs index 8e6c7c31d..eed6a38b9 100644 --- a/eth2/attester/src/test_utils/simulated_beacon_node.rs +++ b/eth2/attester/src/test_utils/simulated_beacon_node.rs @@ -1,6 +1,6 @@ use crate::traits::{BeaconNode, BeaconNodeError, PublishOutcome}; use std::sync::RwLock; -use types::{AttestationData, Attestation, Slot}; +use types::{Attestation, AttestationData, Slot}; type ProduceResult = Result, BeaconNodeError>; type PublishResult = Result; diff --git a/eth2/attester/src/traits.rs b/eth2/attester/src/traits.rs index 1c6d38578..94d138ae0 100644 --- a/eth2/attester/src/traits.rs +++ b/eth2/attester/src/traits.rs @@ -1,4 +1,4 @@ -use types::{AttestationData, Attestation, Signature, Slot}; +use types::{Attestation, AttestationData, Signature, Slot}; #[derive(Debug, PartialEq, Clone)] pub enum BeaconNodeError { diff --git a/eth2/block_proposer/src/lib.rs b/eth2/block_proposer/src/lib.rs index 1d2a4af71..646391c38 100644 --- a/eth2/block_proposer/src/lib.rs +++ b/eth2/block_proposer/src/lib.rs @@ -4,7 +4,7 @@ mod traits; use slot_clock::SlotClock; use ssz::{SignedRoot, TreeHash}; use std::sync::Arc; -use types::{BeaconBlock, ChainSpec, Domain, Slot, Fork}; +use types::{BeaconBlock, ChainSpec, Domain, Fork, Slot}; pub use self::traits::{ BeaconNode, BeaconNodeError, DutiesReader, DutiesReaderError, PublishOutcome, Signer, @@ -57,11 +57,7 @@ pub struct BlockProducer { impl BlockProducer { /// Returns a new instance where `last_processed_slot == 0`. - pub fn new( - spec: Arc, - beacon_node: Arc, - signer: Arc, - ) -> Self { + pub fn new(spec: Arc, beacon_node: Arc, signer: Arc) -> Self { Self { last_processed_slot: None, spec, @@ -72,7 +68,6 @@ impl BlockProducer { } impl BlockProducer { - /* No longer needed because we don't poll any more /// "Poll" to see if the validator is required to take any action. /// @@ -129,7 +124,6 @@ impl BlockProducer { /// The slash-protection code is not yet implemented. There is zero protection against /// slashing. fn produce_block(&mut self, slot: Slot, fork: Fork) -> Result { - let randao_reveal = { // TODO: add domain, etc to this message. Also ensure result matches `into_to_bytes32`. let message = slot.epoch(self.spec.slots_per_epoch).hash_tree_root(); @@ -238,12 +232,8 @@ mod tests { let beacon_node = Arc::new(SimulatedBeaconNode::default()); let signer = Arc::new(LocalSigner::new(Keypair::random())); - - let mut block_proposer = BlockProducer::new( - spec.clone(), - beacon_node.clone(), - signer.clone(), - ); + let mut block_proposer = + BlockProducer::new(spec.clone(), beacon_node.clone(), signer.clone()); // Configure responses from the BeaconNode. beacon_node.set_next_produce_result(Ok(Some(BeaconBlock::random_for_test(&mut rng)))); diff --git a/validator_client/src/attester_service/attestation_grpc_client.rs b/validator_client/src/attester_service/attestation_grpc_client.rs index 97ec05899..05173d824 100644 --- a/validator_client/src/attester_service/attestation_grpc_client.rs +++ b/validator_client/src/attester_service/attestation_grpc_client.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use attester::{BeaconNode, BeaconNodeError, PublishOutcome}; use protos::services::ProduceAttestationDataRequest; -use types::{AttestationData, Attestation, Slot}; +use types::{Attestation, AttestationData, Slot}; pub struct AttestationGrpcClient { client: Arc, diff --git a/validator_client/src/duties/epoch_duties.rs b/validator_client/src/duties/epoch_duties.rs index 984dc6e00..47c9d96c4 100644 --- a/validator_client/src/duties/epoch_duties.rs +++ b/validator_client/src/duties/epoch_duties.rs @@ -44,7 +44,7 @@ impl EpochDuty { slot, shard: self.attestation_shard, committee_index: self.committee_index as usize, - validator_index: self.validator_index as usize + validator_index: self.validator_index as usize, }); } diff --git a/validator_client/src/duties/mod.rs b/validator_client/src/duties/mod.rs index 0b86e3ce9..de08f6ce7 100644 --- a/validator_client/src/duties/mod.rs +++ b/validator_client/src/duties/mod.rs @@ -102,7 +102,7 @@ impl DutiesManager { Ok(Some(work_type)) => current_work.push((validator_signer.clone(), work_type)), Ok(None) => {} // No work for this validator //TODO: This should really log an error, as we shouldn't end up with an err here. - Err(_) => {} // Unknown epoch or validator, no work + Err(_) => {} // Unknown epoch or validator, no work } } if current_work.is_empty() { diff --git a/validator_client/src/service.rs b/validator_client/src/service.rs index 86f310a06..36da36a1f 100644 --- a/validator_client/src/service.rs +++ b/validator_client/src/service.rs @@ -20,6 +20,7 @@ use slog::{debug, error, info, warn}; use slot_clock::{SlotClock, SystemTimeSlotClock}; use std::sync::Arc; use std::sync::RwLock; +use std::thread; use std::time::{Duration, Instant, SystemTime}; use tokio::prelude::*; use tokio::runtime::Builder; @@ -27,7 +28,6 @@ use tokio::timer::Interval; use tokio_timer::clock::Clock; use types::test_utils::generate_deterministic_keypairs; use types::{Epoch, Fork, Slot}; -use std::thread; //TODO: This service should be simplified in the future. Can be made more steamlined. @@ -268,17 +268,11 @@ impl Service { // available AttestationDuty info let attestation_duty = work_type.attestation_duty.expect("Cannot be None"); - let attester_grpc_client = - Arc::new( - AttestationGrpcClient::new( - service.attester_client.clone() - ) - ); + let attester_grpc_client = Arc::new(AttestationGrpcClient::new( + service.attester_client.clone(), + )); let signer = Arc::new(AttesterLocalSigner::new(keypair.clone())); - let attester = - Attester::new( - attester_grpc_client, - signer); + let attester = Attester::new(attester_grpc_client, signer); let mut attester_service = AttesterService { attester, poll_interval_millis: POLL_INTERVAL_MILLIS, From 6c8abd8990e552dc847205edf5b3dc2f33c5fcc9 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Fri, 29 Mar 2019 00:02:41 +1100 Subject: [PATCH 7/8] Fixed merge conflict fail. --- validator_client/src/duties/mod.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/validator_client/src/duties/mod.rs b/validator_client/src/duties/mod.rs index 569a99efa..de08f6ce7 100644 --- a/validator_client/src/duties/mod.rs +++ b/validator_client/src/duties/mod.rs @@ -51,11 +51,7 @@ impl DutiesManager { /// /// be a wall-clock (e.g., system time, remote server time, etc.). fn update(&self, epoch: Epoch) -> Result { -<<<<<<< HEAD let duties = self.beacon_node.request_duties(epoch, &self.signers)?; -======= - let duties = self.beacon_node.request_duties(epoch, &self.pubkeys)?; ->>>>>>> master { // 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) { From be592c86d11fcd6cc0b80e0939d657355ac512b3 Mon Sep 17 00:00:00 2001 From: Luke Anderson Date: Fri, 29 Mar 2019 10:39:37 +1100 Subject: [PATCH 8/8] Started migrating FreeAttestation to Attestation in the harnesses - doesn't compile yet. --- .../test_harness/src/beacon_chain_harness.rs | 12 ++++++------ .../src/validator_harness/direct_beacon_node.rs | 14 +++++++------- .../test_harness/src/validator_harness/mod.rs | 14 ++------------ 3 files changed, 15 insertions(+), 25 deletions(-) diff --git a/beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs b/beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs index 942497476..9f6643c7d 100644 --- a/beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs +++ b/beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs @@ -137,11 +137,11 @@ impl BeaconChainHarness { slot } - /// Gather the `FreeAttestation`s from the valiators. + /// Gather the `Attestation`s from the valiators. /// /// Note: validators will only produce attestations _once per slot_. So, if you call this twice /// you'll only get attestations on the first run. - pub fn gather_free_attesations(&mut self) -> Vec { + pub fn gather_attesations(&mut self) -> Vec { let present_slot = self.beacon_chain.present_slot(); let attesting_validators = self @@ -158,7 +158,7 @@ impl BeaconChainHarness { let attesting_validators: HashSet = HashSet::from_iter(attesting_validators.iter().cloned()); - let free_attestations: Vec = self + let attestations: Vec = self .validators .par_iter_mut() .enumerate() @@ -176,8 +176,8 @@ impl BeaconChainHarness { .collect(); debug!( - "Gathered {} FreeAttestations for slot {}.", - free_attestations.len(), + "Gathered {} Attestations for slot {}.", + attestations.len(), present_slot ); @@ -232,7 +232,7 @@ impl BeaconChainHarness { .unwrap(); }); - debug!("Free attestations processed."); + debug!("attestations processed."); block } diff --git a/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs b/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs index 17630833b..7fc0376c8 100644 --- a/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs +++ b/beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs @@ -12,7 +12,7 @@ use fork_choice::ForkChoice; use parking_lot::RwLock; use slot_clock::SlotClock; use std::sync::Arc; -use types::{AttestationData, BeaconBlock, FreeAttestation, Signature, Slot}; +use types::{AttestationData, BeaconBlock, Attestation, Signature, Slot}; // mod attester; // mod producer; @@ -20,13 +20,13 @@ use types::{AttestationData, BeaconBlock, FreeAttestation, Signature, Slot}; /// Connect directly to a borrowed `BeaconChain` instance so an attester/producer can request/submit /// blocks/attestations. /// -/// `BeaconBlock`s and `FreeAttestation`s are not actually published to the `BeaconChain`, instead +/// `BeaconBlock`s and `Attestation`s are not actually published to the `BeaconChain`, instead /// they are stored inside this struct. This is to allow one to benchmark the submission of the /// block/attestation directly, or modify it before submission. pub struct DirectBeaconNode { beacon_chain: Arc>, published_blocks: RwLock>, - published_attestations: RwLock>, + published_attestations: RwLock>, } impl DirectBeaconNode { @@ -44,7 +44,7 @@ impl DirectBeaconNode { } /// Get the last published attestation (if any). - pub fn last_published_free_attestation(&self) -> Option { + pub fn last_published_free_attestation(&self) -> Option { Some(self.published_attestations.read().last()?.clone()) } } @@ -55,7 +55,7 @@ impl AttesterBeaconNode for DirectBeac _slot: Slot, shard: u64, ) -> Result, NodeError> { - match self.beacon_chain.produce_attestation(shard) { + match self.beacon_chain.produce_attestation_data(shard) { Ok(attestation_data) => Ok(Some(attestation_data)), Err(e) => Err(NodeError::RemoteFailure(format!("{:?}", e))), } @@ -63,9 +63,9 @@ impl AttesterBeaconNode for DirectBeac fn publish_attestation( &self, - free_attestation: FreeAttestation, + attestation: Attestation, ) -> Result { - self.published_attestations.write().push(free_attestation); + self.published_attestations.write().push(attestation); Ok(AttestationPublishOutcome::ValidAttestation) } } diff --git a/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs b/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs index 43ad03ea7..60258c794 100644 --- a/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs +++ b/beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs @@ -2,7 +2,7 @@ mod direct_beacon_node; mod direct_duties; mod local_signer; -use crate::direct_beacon_node::DirectBeaconNode; +use crate::validator_harness::direct_beacon_node::DirectBeaconNode; use attester::PollOutcome as AttestationPollOutcome; use attester::{Attester, Error as AttestationPollError}; use beacon_chain::BeaconChain; @@ -44,10 +44,8 @@ pub struct ValidatorHarness { pub block_producer: TestingBlockProducer, pub attester: TestingAttester, pub spec: Arc, - pub epoch_map: Arc>>, pub keypair: Keypair, pub beacon_node: Arc>>, - pub slot_clock: Arc, pub signer: Arc, } @@ -61,22 +59,16 @@ impl ValidatorHarness { beacon_chain: Arc>>, spec: Arc, ) -> Self { - let slot_clock = Arc::new(TestingSlotClock::new(spec.genesis_slot.as_u64())); let signer = Arc::new(LocalSigner::new(keypair.clone())); let beacon_node = Arc::new(DirectBeaconNode::new(beacon_chain.clone())); - let epoch_map = Arc::new(DirectDuties::new(keypair.pk.clone(), beacon_chain.clone())); let block_producer = BlockProducer::new( spec.clone(), - epoch_map.clone(), - slot_clock.clone(), beacon_node.clone(), signer.clone(), ); let attester = Attester::new( - epoch_map.clone(), - slot_clock.clone(), beacon_node.clone(), signer.clone(), ); @@ -85,10 +77,8 @@ impl ValidatorHarness { block_producer, attester, spec, - epoch_map, keypair, beacon_node, - slot_clock, signer, } } @@ -113,7 +103,7 @@ impl ValidatorHarness { /// Run the `poll` function on the `Attester` and produce a `FreeAttestation`. /// /// An error is returned if the attester refuses to attest. - pub fn produce_free_attestation(&mut self) -> Result { + pub fn produce_attestation(&mut self) -> Result { match self.attester.poll() { Ok(AttestationPollOutcome::AttestationProduced(_)) => {} Ok(outcome) => return Err(AttestationProduceError::DidNotProduce(outcome)),