gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-11 04:31:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Tidy val client

Browse Source
This commit is contained in:
Paul Hauner
2019-11-23 20:42:07 +11:00
parent f62deb7934
commit 7991bbcc22
5 changed files with 187 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
validator_client/src/attestation_service.rs
View File

@@ -3,19 +3,17 @@ use environment::RuntimeContext;
use exit_future::Signal;
use futures::{Future, Stream};
use remote_beacon_node::{PublishStatus, RemoteBeaconNode, ValidatorDuty};
use slog::{error, info, trace};
use slog::{crit, info, trace};
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{ChainSpec, CommitteeIndex, EthSpec, Slot};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
#[derive(Clone)]
pub struct AttestationServiceBuilder<T: Clone, E: EthSpec> {
/// Builds an `AttestationService`.
pub struct AttestationServiceBuilder<T, E: EthSpec> {
duties_service: Option<DutiesService<T, E>>,
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<T>,
@@ -23,8 +21,7 @@ pub struct AttestationServiceBuilder<T: Clone, E: EthSpec> {
context: Option<RuntimeContext<E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationServiceBuilder<T, E> {
impl<T: SlotClock + 'static, E: EthSpec> AttestationServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
duties_service: None,
@@ -83,7 +80,8 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationServiceBuilder<T, E>
}
}
pub struct Inner<T: Clone, E: EthSpec> {
/// Helper to minimise `Arc` usage.
pub struct Inner<T, E: EthSpec> {
duties_service: DutiesService<T, E>,
validator_store: ValidatorStore<T, E>,
slot_clock: T,
@@ -91,19 +89,38 @@ pub struct Inner<T: Clone, E: EthSpec> {
context: RuntimeContext<E>,
}
#[derive(Clone)]
pub struct AttestationService<T: Clone, E: EthSpec> {
/// Attempts to produce attestations for all known validators 1/3rd of the way through each slot.
///
/// If any validators are on the same committee, a single attestation will be downloaded and
/// returned to the beacon node. This attestation will have a signature from each of the
/// validators.
pub struct AttestationService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
impl<T, E: EthSpec> Clone for AttestationService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for AttestationService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
/// Starts the service which periodically produces attestations.
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let context = &self.inner.context;
let context = &self.context;
let log = context.log.clone();
let duration_to_next_slot = self
.inner
.slot_clock
.duration_to_next_slot()
.ok_or_else(|| "Unable to determine duration to next slot".to_string())?;
@@ -111,7 +128,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
let interval = {
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
Interval::new(
Instant::now() + duration_to_next_slot * 3 / 2 + TIME_DELAY_FROM_SLOT,
Instant::now() + duration_to_next_slot + slot_duration / 3,
slot_duration,
)
};
@@ -127,15 +144,15 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
.until(
interval
.map_err(move |e| {
error! {
crit! {
log_1,
"Timer thread failed";
"error" => format!("{}", e)
}
})
.for_each(move |_| {
if let Err(e) = service.clone().spawn_attestation_tasks() {
error!(
if let Err(e) = service.spawn_attestation_tasks() {
crit!(
log_2,
"Failed to spawn attestation tasks";
"error" => e
@@ -158,17 +175,19 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
Ok(exit_signal)
}
/// For each each required attestation, spawn a new task that downloads, signs and uploads the
/// attestation to the beacon node.
fn spawn_attestation_tasks(&self) -> Result<(), String> {
let inner = self.inner.clone();
let service = self.clone();
let slot = inner
let slot = service
.slot_clock
.now()
.ok_or_else(|| "Failed to read slot clock".to_string())?;
let mut committee_indices: HashMap<CommitteeIndex, Vec<ValidatorDuty>> = HashMap::new();
inner
service
.duties_service
.attesters(slot)
.into_iter()
@@ -185,7 +204,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
.into_iter()
.for_each(|(committee_index, validator_duties)| {
// Spawn a separate task for each attestation.
inner.context.executor.spawn(self.clone().do_attestation(
service.context.executor.spawn(self.clone().do_attestation(
slot,
committee_index,
validator_duties,
@@ -195,28 +214,29 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
Ok(())
}
/// For a given `committee_index`, download the attestation, have it signed by all validators
/// in `validator_duties` then upload it.
fn do_attestation(
self,
&self,
slot: Slot,
committee_index: CommitteeIndex,
validator_duties: Vec<ValidatorDuty>,
) -> impl Future<Item = (), Error = ()> {
let inner_1 = self.inner.clone();
let inner_2 = self.inner.clone();
let log_1 = self.inner.context.log.clone();
let log_2 = self.inner.context.log.clone();
let service_1 = self.clone();
let service_2 = self.clone();
let log_1 = self.context.log.clone();
let log_2 = self.context.log.clone();
self.inner
.beacon_node
self.beacon_node
.http
.validator()
.produce_attestation(slot, committee_index)
.map_err(|e| format!("Failed to produce attestation: {:?}", e))
.and_then(move |attestation| {
.map(move |attestation| {
validator_duties
.iter()
.try_fold(attestation, |attestation, duty| {
let log = inner_1.context.log.clone();
.fold(attestation, |mut attestation, duty| {
let log = service_1.context.log.clone();
if let Some((
duty_slot,
@@ -225,28 +245,29 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
)) = attestation_duties(duty)
{
if duty_slot == slot && duty_committee_index == committee_index {
inner_1
if service_1
.validator_store
.sign_attestation(
&duty.validator_pubkey,
validator_committee_position,
attestation,
&mut attestation,
)
.ok_or_else(|| "Unable to sign attestation".to_string())
.is_none()
{
crit!(log, "Failed to sign attestation");
}
} else {
error!(log, "Inconsistent validator duties during signing");
Ok(attestation)
crit!(log, "Inconsistent validator duties during signing");
}
} else {
error!(log, "Missing validator duties when signing");
Ok(attestation)
crit!(log, "Missing validator duties when signing");
}
attestation
})
})
.and_then(move |attestation| {
inner_2
service_2
.beacon_node
.http
.validator()
@@ -263,7 +284,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
"committee_index" => attestation.data.index,
"slot" => attestation.data.slot.as_u64(),
),
PublishStatus::Invalid(msg) => error!(
PublishStatus::Invalid(msg) => crit!(
log_1,
"Published attestation was invalid";
"message" => msg,
@@ -271,11 +292,11 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
"slot" => attestation.data.slot.as_u64(),
),
PublishStatus::Unknown => {
info!(log_1, "Unknown condition when publishing attestation")
crit!(log_1, "Unknown condition when publishing attestation")
}
})
.map_err(move |e| {
error!(
crit!(
log_2,
"Error during attestation production";
"error" => e
@@ -284,7 +305,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> AttestationService<T, E> {
}
}
pub fn attestation_duties(duty: &ValidatorDuty) -> Option<(Slot, CommitteeIndex, usize)> {
fn attestation_duties(duty: &ValidatorDuty) -> Option<(Slot, CommitteeIndex, usize)> {
Some((
duty.attestation_slot?,
duty.attestation_committee_index?,

66
validator_client/src/block_service.rs
View File

@@ -5,6 +5,7 @@ use futures::{stream, Future, IntoFuture, Stream};
use remote_beacon_node::{PublishStatus, RemoteBeaconNode};
use slog::{error, info};
use slot_clock::SlotClock;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
@@ -13,8 +14,8 @@ use types::{ChainSpec, EthSpec};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
#[derive(Clone)]
pub struct BlockServiceBuilder<T: Clone, E: EthSpec> {
/// Builds a `BlockService`.
pub struct BlockServiceBuilder<T, E: EthSpec> {
duties_service: Option<DutiesService<T, E>>,
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<Arc<T>>,
@@ -22,8 +23,7 @@ pub struct BlockServiceBuilder<T: Clone, E: EthSpec> {
context: Option<RuntimeContext<E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
duties_service: None,
@@ -61,27 +61,28 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
pub fn build(self) -> Result<BlockService<T, E>, String> {
Ok(BlockService {
duties_service: self
.duties_service
.ok_or_else(|| "Cannot build BlockService without duties_service")?,
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build BlockService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build BlockService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build BlockService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build BlockService without runtime_context")?,
inner: Arc::new(Inner {
duties_service: self
.duties_service
.ok_or_else(|| "Cannot build BlockService without duties_service")?,
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build BlockService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build BlockService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build BlockService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build BlockService without runtime_context")?,
}),
})
}
}
#[derive(Clone)]
pub struct BlockService<T: Clone, E: EthSpec> {
pub struct Inner<T, E: EthSpec> {
duties_service: DutiesService<T, E>,
validator_store: ValidatorStore<T, E>,
slot_clock: Arc<T>,
@@ -89,8 +90,27 @@ pub struct BlockService<T: Clone, E: EthSpec> {
context: RuntimeContext<E>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> BlockService<T, E> {
pub struct BlockService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for BlockService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for BlockService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let log = self.context.log.clone();

70
validator_client/src/duties_service.rs
View File

@@ -7,6 +7,7 @@ use remote_beacon_node::{RemoteBeaconNode, ValidatorDuty};
use slog::{error, info, trace, warn};
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
@@ -121,20 +122,16 @@ impl DutiesStore {
}
}
#[derive(Clone)]
pub struct DutiesServiceBuilder<T: Clone, E: EthSpec> {
store: Option<Arc<DutiesStore>>,
pub struct DutiesServiceBuilder<T, E: EthSpec> {
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<Arc<T>>,
slot_clock: Option<T>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> DutiesServiceBuilder<T, E> {
impl<T: SlotClock + 'static, E: EthSpec> DutiesServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
store: None,
validator_store: None,
slot_clock: None,
beacon_node: None,
@@ -148,7 +145,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> DutiesServiceBuilder<T, E> {
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(Arc::new(slot_clock));
self.slot_clock = Some(slot_clock);
self
}
@@ -164,33 +161,54 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> DutiesServiceBuilder<T, E> {
pub fn build(self) -> Result<DutiesService<T, E>, String> {
Ok(DutiesService {
store: Arc::new(DutiesStore::default()),
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build DutiesService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build DutiesService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build DutiesService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build DutiesService without runtime_context")?,
inner: Arc::new(Inner {
store: Arc::new(DutiesStore::default()),
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build DutiesService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build DutiesService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build DutiesService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build DutiesService without runtime_context")?,
}),
})
}
}
#[derive(Clone)]
pub struct DutiesService<T: Clone, E: EthSpec> {
pub struct Inner<T, E: EthSpec> {
store: Arc<DutiesStore>,
validator_store: ValidatorStore<T, E>,
slot_clock: Arc<T>,
slot_clock: T,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
}
impl<T: SlotClock + Clone + 'static, E: EthSpec> DutiesService<T, E> {
pub struct DutiesService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for DutiesService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for DutiesService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
/// Returns the pubkeys of the validators which are assigned to propose in the given slot.
///
/// In normal cases, there should be 0 or 1 validators returned. In extreme cases (i.e., deep forking)
@@ -251,7 +269,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> DutiesService<T, E> {
Ok(exit_signal)
}
fn do_update(self) -> impl Future<Item = (), Error = ()> {
fn do_update(&self) -> impl Future<Item = (), Error = ()> {
let service_1 = self.clone();
let service_2 = self.clone();
let service_3 = self.clone();

47
validator_client/src/fork_service.rs
View File

@@ -5,6 +5,7 @@ use parking_lot::RwLock;
use remote_beacon_node::RemoteBeaconNode;
use slog::{error, info, trace};
use slot_clock::SlotClock;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
@@ -13,16 +14,14 @@ use types::{ChainSpec, EthSpec, Fork};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(80);
#[derive(Clone)]
pub struct ForkServiceBuilder<T: Clone, E: EthSpec> {
pub struct ForkServiceBuilder<T, E: EthSpec> {
fork: Option<Fork>,
slot_clock: Option<T>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkServiceBuilder<T, E> {
impl<T: SlotClock + 'static, E: EthSpec> ForkServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
fork: None,
@@ -65,29 +64,42 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkServiceBuilder<T, E> {
}
}
struct Inner<T, E: EthSpec> {
pub struct Inner<T, E: EthSpec> {
fork: RwLock<Option<Fork>>,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
slot_clock: T,
}
#[derive(Clone)]
pub struct ForkService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
// TODO: clean trait bounds.
impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkService<T, E> {
impl<T, E: EthSpec> Clone for ForkService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for ForkService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> ForkService<T, E> {
pub fn fork(&self) -> Option<Fork> {
self.inner.fork.read().clone()
self.fork.read().clone()
}
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let log = self.inner.context.log.clone();
let log = self.context.log.clone();
let duration_to_next_epoch = self
.inner
.slot_clock
.duration_to_next_epoch(E::slots_per_epoch())
.ok_or_else(|| "Unable to determine duration to next epoch".to_string())?;
@@ -106,12 +118,9 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkService<T, E> {
let log_2 = log.clone();
// Run an immediate update before starting the updater service.
self.inner
.context
.executor
.spawn(service.clone().do_update());
self.context.executor.spawn(service.clone().do_update());
self.inner.context.executor.spawn(
self.context.executor.spawn(
exit_fut
.until(
interval
@@ -122,7 +131,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkService<T, E> {
"error" => format!("{}", e)
}
})
.for_each(move |_| service.clone().do_update())
.for_each(move |_| service.do_update())
// Prevent any errors from escaping and stopping the interval.
.then(|_| Ok(())),
)
@@ -132,8 +141,8 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ForkService<T, E> {
Ok(exit_signal)
}
fn do_update(self) -> impl Future<Item = (), Error = ()> {
let service_1 = self.inner.clone();
fn do_update(&self) -> impl Future<Item = (), Error = ()> {
let service_1 = self.clone();
let log_1 = service_1.context.log.clone();
let log_2 = service_1.context.log.clone();

8
validator_client/src/validator_store.rs
View File

@@ -27,7 +27,7 @@ pub struct ValidatorStore<T, E: EthSpec> {
_phantom: PhantomData<E>,
}
impl<T: SlotClock + Clone + 'static, E: EthSpec> ValidatorStore<T, E> {
impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
pub fn load_from_disk(
base_dir: PathBuf,
spec: ChainSpec,
@@ -171,8 +171,8 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ValidatorStore<T, E> {
&self,
validator_pubkey: &PublicKey,
validator_committee_position: usize,
mut attestation: Attestation<E>,
) -> Option<Attestation<E>> {
attestation: &mut Attestation<E>,
) -> Option<()> {
// TODO: check for slashing.
self.validators
.read()
@@ -196,7 +196,7 @@ impl<T: SlotClock + Clone + 'static, E: EthSpec> ValidatorStore<T, E> {
})
.ok()?;
Some(attestation)
Some(())
})
}
}