mirror of
https://github.com/sigp/lighthouse.git
synced 2026-03-03 00:31:50 +00:00
Add attestation gossip pre-verification (#983)
* Add PH & MS slot clock changes * Account for genesis time * Add progress on duties refactor * Add simple is_aggregator bool to val subscription * Start work on attestation_verification.rs * Add progress on ObservedAttestations * Progress with ObservedAttestations * Fix tests * Add observed attestations to the beacon chain * Add attestation observation to processing code * Add progress on attestation verification * Add first draft of ObservedAttesters * Add more tests * Add observed attesters to beacon chain * Add observers to attestation processing * Add more attestation verification * Create ObservedAggregators map * Remove commented-out code * Add observed aggregators into chain * Add progress * Finish adding features to attestation verification * Ensure beacon chain compiles * Link attn verification into chain * Integrate new attn verification in chain * Remove old attestation processing code * Start trying to fix beacon_chain tests * Split adding into pools into two functions * Add aggregation to harness * Get test harness working again * Adjust the number of aggregators for test harness * Fix edge-case in harness * Integrate new attn processing in network * Fix compile bug in validator_client * Update validator API endpoints * Fix aggreagation in test harness * Fix enum thing * Fix attestation observation bug: * Patch failing API tests * Start adding comments to attestation verification * Remove unused attestation field * Unify "is block known" logic * Update comments * Supress fork choice errors for network processing * Add todos * Tidy * Add gossip attn tests * Disallow test harness to produce old attns * Comment out in-progress tests * Partially address pruning tests * Fix failing store test * Add aggregate tests * Add comments about which spec conditions we check * Dont re-aggregate * Split apart test harness attn production * Fix compile error in network * Make progress on commented-out test * Fix skipping attestation test * Add fork choice verification tests * Tidy attn tests, remove dead code * Remove some accidentally added code * Fix clippy lint * Rename test file * Add block tests, add cheap block proposer check * Rename block testing file * Add observed_block_producers * Tidy * Switch around block signature verification * Finish block testing * Remove gossip from signature tests * First pass of self review * Fix deviation in spec * Update test spec tags * Start moving over to hashset * Finish moving observed attesters to hashmap * Move aggregation pool over to hashmap * Make fc attn borrow again * Fix rest_api compile error * Fix missing comments * Fix monster test * Uncomment increasing slots test * Address remaining comments * Remove unsafe, use cfg test * Remove cfg test flag * Fix dodgy comment * Ignore aggregates that are already known. * Unify aggregator modulo logic * Fix typo in logs * Refactor validator subscription logic * Avoid reproducing selection proof * Skip HTTP call if no subscriptions * Rename DutyAndState -> DutyAndProof * Tidy logs * Print root as dbg * Fix compile errors in tests * Fix compile error in test
This commit is contained in:
Paul Hauner
@@ -1,12 +1,11 @@
|
||||
use crate::{
|
||||
duties_service::{DutiesService, DutyAndState},
|
||||
duties_service::{DutiesService, DutyAndProof},
|
||||
validator_store::ValidatorStore,
|
||||
};
|
||||
use environment::RuntimeContext;
|
||||
use exit_future::Signal;
|
||||
use futures::{future, Future, Stream};
|
||||
use remote_beacon_node::{PublishStatus, RemoteBeaconNode};
|
||||
use rest_types::ValidatorSubscription;
|
||||
use slog::{crit, debug, info, trace};
|
||||
use slot_clock::SlotClock;
|
||||
use std::collections::HashMap;
|
||||
@@ -198,31 +197,15 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
.checked_sub(slot_duration / 3)
|
||||
.unwrap_or_else(|| Duration::from_secs(0));
|
||||
|
||||
let epoch = slot.epoch(E::slots_per_epoch());
|
||||
// Check if any attestation subscriptions are required. If there a new attestation duties for
|
||||
// this epoch or the next, send them to the beacon node
|
||||
let mut duties_to_subscribe = service.duties_service.unsubscribed_epoch_duties(&epoch);
|
||||
duties_to_subscribe.append(
|
||||
&mut service
|
||||
.duties_service
|
||||
.unsubscribed_epoch_duties(&(epoch + 1)),
|
||||
);
|
||||
|
||||
// spawn a task to subscribe all the duties
|
||||
service
|
||||
.context
|
||||
.executor
|
||||
.spawn(self.clone().send_subscriptions(duties_to_subscribe));
|
||||
|
||||
let duties_by_committee_index: HashMap<CommitteeIndex, Vec<DutyAndState>> = service
|
||||
let duties_by_committee_index: HashMap<CommitteeIndex, Vec<DutyAndProof>> = service
|
||||
.duties_service
|
||||
.attesters(slot)
|
||||
.into_iter()
|
||||
.fold(HashMap::new(), |mut map, duty_and_state| {
|
||||
if let Some(committee_index) = duty_and_state.duty.attestation_committee_index {
|
||||
.fold(HashMap::new(), |mut map, duty_and_proof| {
|
||||
if let Some(committee_index) = duty_and_proof.duty.attestation_committee_index {
|
||||
let validator_duties = map.entry(committee_index).or_insert_with(|| vec![]);
|
||||
|
||||
validator_duties.push(duty_and_state);
|
||||
validator_duties.push(duty_and_proof);
|
||||
}
|
||||
|
||||
map
|
||||
@@ -250,81 +233,6 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Subscribes any required validators to the beacon node for a particular slot.
|
||||
///
|
||||
/// This informs the beacon node that the validator has a duty on a particular
|
||||
/// slot allowing the beacon node to connect to the required subnet and determine
|
||||
/// if attestations need to be aggregated.
|
||||
fn send_subscriptions(&self, duties: Vec<DutyAndState>) -> impl Future<Item = (), Error = ()> {
|
||||
let service_1 = self.clone();
|
||||
let num_duties = duties.len();
|
||||
|
||||
let log_1 = self.context.log.clone();
|
||||
let log_2 = self.context.log.clone();
|
||||
|
||||
let (validator_subscriptions, successful_duties): (Vec<_>, Vec<_>) = duties
|
||||
.into_iter()
|
||||
.filter_map(|duty| {
|
||||
let (slot, attestation_committee_index, _, validator_index) =
|
||||
duty.attestation_duties()?;
|
||||
let selection_proof = self
|
||||
.validator_store
|
||||
.produce_selection_proof(duty.validator_pubkey(), slot)?;
|
||||
let modulo = duty.duty.aggregator_modulo?;
|
||||
let subscription = ValidatorSubscription {
|
||||
validator_index,
|
||||
attestation_committee_index,
|
||||
slot,
|
||||
is_aggregator: selection_proof
|
||||
.is_aggregator(modulo)
|
||||
.map_err(|e| crit!(log_1, "Unable to determine aggregator: {:?}", e))
|
||||
.ok()?,
|
||||
};
|
||||
|
||||
Some((subscription, (duty, selection_proof)))
|
||||
})
|
||||
.unzip();
|
||||
|
||||
let num_failed_duties = num_duties - successful_duties.len();
|
||||
|
||||
self.beacon_node
|
||||
.http
|
||||
.validator()
|
||||
.subscribe(validator_subscriptions)
|
||||
.map_err(|e| format!("Failed to subscribe validators: {:?}", e))
|
||||
.map(move |publish_status| match publish_status {
|
||||
PublishStatus::Valid => info!(
|
||||
log_1,
|
||||
"Successfully subscribed validators";
|
||||
"validators" => num_duties,
|
||||
"failed_validators" => num_failed_duties,
|
||||
),
|
||||
PublishStatus::Invalid(msg) => crit!(
|
||||
log_1,
|
||||
"Validator Subscription was invalid";
|
||||
"message" => msg,
|
||||
),
|
||||
PublishStatus::Unknown => {
|
||||
crit!(log_1, "Unknown condition when publishing attestation")
|
||||
}
|
||||
})
|
||||
.and_then(move |_| {
|
||||
for (duty, selection_proof) in successful_duties {
|
||||
service_1
|
||||
.duties_service
|
||||
.subscribe_duty(&duty.duty, selection_proof);
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.map_err(move |e| {
|
||||
crit!(
|
||||
log_2,
|
||||
"Error during attestation production";
|
||||
"error" => e
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// Performs the first step of the attesting process: downloading `Attestation` objects,
|
||||
/// signing them and returning them to the validator.
|
||||
///
|
||||
@@ -338,7 +246,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
&self,
|
||||
slot: Slot,
|
||||
committee_index: CommitteeIndex,
|
||||
validator_duties: Vec<DutyAndState>,
|
||||
validator_duties: Vec<DutyAndProof>,
|
||||
aggregate_production_instant: Instant,
|
||||
) -> Box<dyn Future<Item = (), Error = ()> + Send> {
|
||||
// There's not need to produce `Attestation` or `SignedAggregateAndProof` if we do not have
|
||||
@@ -421,7 +329,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
&self,
|
||||
slot: Slot,
|
||||
committee_index: CommitteeIndex,
|
||||
validator_duties: Arc<Vec<DutyAndState>>,
|
||||
validator_duties: Arc<Vec<DutyAndProof>>,
|
||||
) -> Box<dyn Future<Item = Option<Attestation<E>>, Error = String> + Send> {
|
||||
if validator_duties.is_empty() {
|
||||
return Box::new(future::ok(None));
|
||||
@@ -522,6 +430,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
"head_block" => format!("{:?}", beacon_block_root),
|
||||
"committee_index" => committee_index,
|
||||
"slot" => slot.as_u64(),
|
||||
"type" => "unaggregated",
|
||||
),
|
||||
PublishStatus::Invalid(msg) => crit!(
|
||||
log,
|
||||
@@ -529,10 +438,12 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
"message" => msg,
|
||||
"committee_index" => committee_index,
|
||||
"slot" => slot.as_u64(),
|
||||
"type" => "unaggregated",
|
||||
),
|
||||
PublishStatus::Unknown => crit!(
|
||||
log,
|
||||
"Unknown condition when publishing unagg. attestation"
|
||||
),
|
||||
PublishStatus::Unknown => {
|
||||
crit!(log, "Unknown condition when publishing attestation")
|
||||
}
|
||||
})
|
||||
.map(|()| Some(attestation)),
|
||||
)
|
||||
@@ -565,7 +476,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
fn produce_and_publish_aggregates(
|
||||
&self,
|
||||
attestation: Attestation<E>,
|
||||
validator_duties: Arc<Vec<DutyAndState>>,
|
||||
validator_duties: Arc<Vec<DutyAndProof>>,
|
||||
) -> impl Future<Item = (), Error = String> {
|
||||
let service_1 = self.clone();
|
||||
let log_1 = self.context.log.clone();
|
||||
@@ -581,23 +492,18 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
// a `SignedAggregateAndProof`
|
||||
let signed_aggregate_and_proofs = validator_duties
|
||||
.iter()
|
||||
.filter_map(|duty_and_state| {
|
||||
.filter_map(|duty_and_proof| {
|
||||
// Do not produce a signed aggregator for validators that are not
|
||||
// subscribed aggregators.
|
||||
//
|
||||
// Note: this function returns `false` if the validator is required to
|
||||
// be an aggregator but has not yet subscribed.
|
||||
if !duty_and_state.is_aggregator() {
|
||||
return None;
|
||||
}
|
||||
let selection_proof = duty_and_proof.selection_proof.as_ref()?.clone();
|
||||
|
||||
let (duty_slot, duty_committee_index, _, validator_index) =
|
||||
duty_and_state.attestation_duties().or_else(|| {
|
||||
duty_and_proof.attestation_duties().or_else(|| {
|
||||
crit!(log_1, "Missing duties when signing aggregate");
|
||||
None
|
||||
})?;
|
||||
|
||||
let pubkey = &duty_and_state.duty.validator_pubkey;
|
||||
let pubkey = &duty_and_proof.duty.validator_pubkey;
|
||||
let slot = attestation.data.slot;
|
||||
let committee_index = attestation.data.index;
|
||||
|
||||
@@ -612,6 +518,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
pubkey,
|
||||
validator_index,
|
||||
aggregated_attestation.clone(),
|
||||
selection_proof,
|
||||
)
|
||||
{
|
||||
Some(signed_aggregate_and_proof)
|
||||
@@ -637,11 +544,12 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
.map(move |(attestation, publish_status)| match publish_status {
|
||||
PublishStatus::Valid => info!(
|
||||
log_1,
|
||||
"Successfully published aggregate attestations";
|
||||
"Successfully published attestations";
|
||||
"signatures" => attestation.aggregation_bits.num_set_bits(),
|
||||
"head_block" => format!("{}", attestation.data.beacon_block_root),
|
||||
"head_block" => format!("{:?}", attestation.data.beacon_block_root),
|
||||
"committee_index" => attestation.data.index,
|
||||
"slot" => attestation.data.slot.as_u64(),
|
||||
"type" => "aggregated",
|
||||
),
|
||||
PublishStatus::Invalid(msg) => crit!(
|
||||
log_1,
|
||||
@@ -649,9 +557,10 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
|
||||
"message" => msg,
|
||||
"committee_index" => attestation.data.index,
|
||||
"slot" => attestation.data.slot.as_u64(),
|
||||
"type" => "aggregated",
|
||||
),
|
||||
PublishStatus::Unknown => {
|
||||
crit!(log_1, "Unknown condition when publishing attestation")
|
||||
crit!(log_1, "Unknown condition when publishing agg. attestation")
|
||||
}
|
||||
}))
|
||||
} else {
|
||||
|
||||
@@ -3,8 +3,8 @@ use environment::RuntimeContext;
|
||||
use exit_future::Signal;
|
||||
use futures::{future, Future, IntoFuture, Stream};
|
||||
use parking_lot::RwLock;
|
||||
use remote_beacon_node::RemoteBeaconNode;
|
||||
use rest_types::{ValidatorDuty, ValidatorDutyBytes};
|
||||
use remote_beacon_node::{PublishStatus, RemoteBeaconNode};
|
||||
use rest_types::{ValidatorDuty, ValidatorDutyBytes, ValidatorSubscription};
|
||||
use slog::{crit, debug, error, info, trace, warn};
|
||||
use slot_clock::SlotClock;
|
||||
use std::collections::HashMap;
|
||||
@@ -21,49 +21,73 @@ const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
|
||||
/// Remove any duties where the `duties_epoch < current_epoch - PRUNE_DEPTH`.
|
||||
const PRUNE_DEPTH: u64 = 4;
|
||||
|
||||
type BaseHashMap = HashMap<PublicKey, HashMap<Epoch, DutyAndState>>;
|
||||
type BaseHashMap = HashMap<PublicKey, HashMap<Epoch, DutyAndProof>>;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum DutyState {
|
||||
/// This duty has not been subscribed to the beacon node.
|
||||
NotSubscribed,
|
||||
/// The duty has been subscribed and the validator is an aggregator for this duty. The
|
||||
/// selection proof is provided to construct the `AggregateAndProof` struct.
|
||||
SubscribedAggregator(SelectionProof),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct DutyAndState {
|
||||
pub struct DutyAndProof {
|
||||
/// The validator duty.
|
||||
pub duty: ValidatorDuty,
|
||||
/// The current state of the validator duty.
|
||||
state: DutyState,
|
||||
/// Stores the selection proof if the duty elects the validator to be an aggregator.
|
||||
pub selection_proof: Option<SelectionProof>,
|
||||
}
|
||||
|
||||
impl DutyAndState {
|
||||
/// Returns true if the duty is an aggregation duty (the validator must aggregate all
|
||||
/// attestations.
|
||||
pub fn is_aggregator(&self) -> bool {
|
||||
match self.state {
|
||||
DutyState::NotSubscribed => false,
|
||||
DutyState::SubscribedAggregator(_) => true,
|
||||
}
|
||||
impl DutyAndProof {
|
||||
/// Computes the selection proof for `self.validator_pubkey` and `self.duty.attestation_slot`,
|
||||
/// storing it in `self.selection_proof` _if_ the validator is an aggregator. If the validator
|
||||
/// is not an aggregator, `self.selection_proof` is set to `None`.
|
||||
///
|
||||
/// ## Errors
|
||||
///
|
||||
/// - `self.validator_pubkey` is not known in `validator_store`.
|
||||
/// - There's an arith error during computation.
|
||||
pub fn compute_selection_proof<T: SlotClock + 'static, E: EthSpec>(
|
||||
&mut self,
|
||||
validator_store: &ValidatorStore<T, E>,
|
||||
) -> Result<(), String> {
|
||||
let (modulo, slot) = if let (Some(modulo), Some(slot)) =
|
||||
(self.duty.aggregator_modulo, self.duty.attestation_slot)
|
||||
{
|
||||
(modulo, slot)
|
||||
} else {
|
||||
// If there is no modulo or for the aggregator we assume they are not activated and
|
||||
// therefore not an aggregator.
|
||||
self.selection_proof = None;
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let selection_proof = validator_store
|
||||
.produce_selection_proof(&self.duty.validator_pubkey, slot)
|
||||
.ok_or_else(|| "Validator pubkey missing from store".to_string())?;
|
||||
|
||||
self.selection_proof = selection_proof
|
||||
.is_aggregator_from_modulo(modulo)
|
||||
.map_err(|e| format!("Invalid modulo: {:?}", e))
|
||||
.map(|is_aggregator| {
|
||||
if is_aggregator {
|
||||
Some(selection_proof)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns the selection proof if the duty is an aggregation duty.
|
||||
pub fn selection_proof(&self) -> Option<SelectionProof> {
|
||||
match &self.state {
|
||||
DutyState::SubscribedAggregator(proof) => Some(proof.clone()),
|
||||
_ => None,
|
||||
}
|
||||
/// Returns `true` if the two `Self` instances would result in the same beacon subscription.
|
||||
pub fn subscription_eq(&self, other: &Self) -> bool {
|
||||
self.selection_proof_eq(other)
|
||||
&& self.duty.validator_index == other.duty.validator_index
|
||||
&& self.duty.attestation_committee_index == other.duty.attestation_committee_index
|
||||
&& self.duty.attestation_slot == other.duty.attestation_slot
|
||||
}
|
||||
|
||||
/// Returns true if the this duty has been subscribed with the beacon node.
|
||||
pub fn is_subscribed(&self) -> bool {
|
||||
match self.state {
|
||||
DutyState::NotSubscribed => false,
|
||||
DutyState::SubscribedAggregator(_) => true,
|
||||
}
|
||||
/// Returns `true` if the selection proof between `self` and `other` _should_ be equal.
|
||||
///
|
||||
/// It's important to note that this doesn't actually check `self.selection_proof`, instead it
|
||||
/// checks to see if the inputs to computing the selection proof are equal.
|
||||
fn selection_proof_eq(&self, other: &Self) -> bool {
|
||||
self.duty.aggregator_modulo == other.duty.aggregator_modulo
|
||||
&& self.duty.attestation_slot == other.duty.attestation_slot
|
||||
}
|
||||
|
||||
/// Returns the information required for an attesting validator, if they are scheduled to
|
||||
@@ -82,10 +106,10 @@ impl DutyAndState {
|
||||
}
|
||||
}
|
||||
|
||||
impl TryInto<DutyAndState> for ValidatorDutyBytes {
|
||||
impl TryInto<DutyAndProof> for ValidatorDutyBytes {
|
||||
type Error = String;
|
||||
|
||||
fn try_into(self) -> Result<DutyAndState, Self::Error> {
|
||||
fn try_into(self) -> Result<DutyAndProof, Self::Error> {
|
||||
let duty = ValidatorDuty {
|
||||
validator_pubkey: (&self.validator_pubkey)
|
||||
.try_into()
|
||||
@@ -97,9 +121,9 @@ impl TryInto<DutyAndState> for ValidatorDutyBytes {
|
||||
block_proposal_slots: self.block_proposal_slots,
|
||||
aggregator_modulo: self.aggregator_modulo,
|
||||
};
|
||||
Ok(DutyAndState {
|
||||
Ok(DutyAndProof {
|
||||
duty,
|
||||
state: DutyState::NotSubscribed,
|
||||
selection_proof: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -114,11 +138,24 @@ enum InsertOutcome {
|
||||
Identical,
|
||||
/// There were duties for this validator and epoch in the store that were different to the ones
|
||||
/// provided. The existing duties were replaced.
|
||||
Replaced,
|
||||
Replaced { should_resubscribe: bool },
|
||||
/// The given duties were invalid.
|
||||
Invalid,
|
||||
}
|
||||
|
||||
impl InsertOutcome {
|
||||
/// Returns `true` if the outcome indicates that the validator _might_ require a subscription.
|
||||
pub fn is_subscription_candidate(self) -> bool {
|
||||
match self {
|
||||
InsertOutcome::Replaced { should_resubscribe } => should_resubscribe,
|
||||
InsertOutcome::NewValidator => true,
|
||||
InsertOutcome::NewEpoch => true,
|
||||
InsertOutcome::Identical => false,
|
||||
InsertOutcome::Invalid => false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct DutiesStore {
|
||||
store: RwLock<BaseHashMap>,
|
||||
@@ -173,49 +210,7 @@ impl DutiesStore {
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Gets a list of validator duties for an epoch that have not yet been subscribed
|
||||
/// to the beacon node.
|
||||
// Note: Potentially we should modify the data structure to store the unsubscribed epoch duties for validator clients with a large number of validators. This currently adds an O(N) search each slot.
|
||||
fn unsubscribed_epoch_duties(&self, epoch: &Epoch) -> Vec<DutyAndState> {
|
||||
self.store
|
||||
.read()
|
||||
.iter()
|
||||
.filter_map(|(_validator_pubkey, validator_map)| {
|
||||
validator_map.get(epoch).and_then(|duty_and_state| {
|
||||
if !duty_and_state.is_subscribed() {
|
||||
Some(duty_and_state)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
})
|
||||
.cloned()
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Marks a duty as being subscribed to the beacon node. This is called by the attestation
|
||||
/// service once it has been sent.
|
||||
fn set_duty_state(
|
||||
&self,
|
||||
validator: &PublicKey,
|
||||
slot: Slot,
|
||||
state: DutyState,
|
||||
slots_per_epoch: u64,
|
||||
) {
|
||||
let epoch = slot.epoch(slots_per_epoch);
|
||||
|
||||
let mut store = self.store.write();
|
||||
if let Some(map) = store.get_mut(validator) {
|
||||
if let Some(duty) = map.get_mut(&epoch) {
|
||||
if duty.duty.attestation_slot == Some(slot) {
|
||||
// set the duty state
|
||||
duty.state = state;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn attesters(&self, slot: Slot, slots_per_epoch: u64) -> Vec<DutyAndState> {
|
||||
fn attesters(&self, slot: Slot, slots_per_epoch: u64) -> Vec<DutyAndProof> {
|
||||
self.store
|
||||
.read()
|
||||
.iter()
|
||||
@@ -236,27 +231,49 @@ impl DutiesStore {
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn insert(&self, epoch: Epoch, duties: DutyAndState, slots_per_epoch: u64) -> InsertOutcome {
|
||||
fn insert<T: SlotClock + 'static, E: EthSpec>(
|
||||
&self,
|
||||
epoch: Epoch,
|
||||
mut duties: DutyAndProof,
|
||||
slots_per_epoch: u64,
|
||||
validator_store: &ValidatorStore<T, E>,
|
||||
) -> Result<InsertOutcome, String> {
|
||||
let mut store = self.store.write();
|
||||
|
||||
if !duties_match_epoch(&duties.duty, epoch, slots_per_epoch) {
|
||||
return InsertOutcome::Invalid;
|
||||
return Ok(InsertOutcome::Invalid);
|
||||
}
|
||||
|
||||
// TODO: refactor with Entry.
|
||||
|
||||
if let Some(validator_map) = store.get_mut(&duties.duty.validator_pubkey) {
|
||||
if let Some(known_duties) = validator_map.get_mut(&epoch) {
|
||||
if known_duties.duty == duties.duty {
|
||||
InsertOutcome::Identical
|
||||
Ok(InsertOutcome::Identical)
|
||||
} else {
|
||||
// Compute the selection proof.
|
||||
duties.compute_selection_proof(validator_store)?;
|
||||
|
||||
// Determine if a re-subscription is required.
|
||||
let should_resubscribe = duties.subscription_eq(known_duties);
|
||||
|
||||
// Replace the existing duties.
|
||||
*known_duties = duties;
|
||||
InsertOutcome::Replaced
|
||||
|
||||
Ok(InsertOutcome::Replaced { should_resubscribe })
|
||||
}
|
||||
} else {
|
||||
// Compute the selection proof.
|
||||
duties.compute_selection_proof(validator_store)?;
|
||||
|
||||
validator_map.insert(epoch, duties);
|
||||
|
||||
InsertOutcome::NewEpoch
|
||||
Ok(InsertOutcome::NewEpoch)
|
||||
}
|
||||
} else {
|
||||
// Compute the selection proof.
|
||||
duties.compute_selection_proof(validator_store)?;
|
||||
|
||||
let validator_pubkey = duties.duty.validator_pubkey.clone();
|
||||
|
||||
let mut validator_map = HashMap::new();
|
||||
@@ -264,7 +281,7 @@ impl DutiesStore {
|
||||
|
||||
store.insert(validator_pubkey, validator_map);
|
||||
|
||||
InsertOutcome::NewValidator
|
||||
Ok(InsertOutcome::NewValidator)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -408,29 +425,10 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
|
||||
}
|
||||
|
||||
/// Returns all `ValidatorDuty` for the given `slot`.
|
||||
pub fn attesters(&self, slot: Slot) -> Vec<DutyAndState> {
|
||||
pub fn attesters(&self, slot: Slot) -> Vec<DutyAndProof> {
|
||||
self.store.attesters(slot, E::slots_per_epoch())
|
||||
}
|
||||
|
||||
/// Returns all `ValidatorDuty` that have not been registered with the beacon node.
|
||||
pub fn unsubscribed_epoch_duties(&self, epoch: &Epoch) -> Vec<DutyAndState> {
|
||||
self.store.unsubscribed_epoch_duties(epoch)
|
||||
}
|
||||
|
||||
/// Marks the duty as being subscribed to the beacon node.
|
||||
///
|
||||
/// If the duty is to be marked as an aggregator duty, a selection proof is also provided.
|
||||
pub fn subscribe_duty(&self, duty: &ValidatorDuty, proof: SelectionProof) {
|
||||
if let Some(slot) = duty.attestation_slot {
|
||||
self.store.set_duty_state(
|
||||
&duty.validator_pubkey,
|
||||
slot,
|
||||
DutyState::SubscribedAggregator(proof),
|
||||
E::slots_per_epoch(),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the service that periodically polls the beacon node for validator duties.
|
||||
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
|
||||
let log = self.context.log.clone();
|
||||
@@ -569,7 +567,8 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
|
||||
/// Attempt to download the duties of all managed validators for the given `epoch`.
|
||||
fn update_epoch(self, epoch: Epoch) -> impl Future<Item = (), Error = String> {
|
||||
let service_1 = self.clone();
|
||||
let service_2 = self;
|
||||
let service_2 = self.clone();
|
||||
let service_3 = self;
|
||||
|
||||
let pubkeys = service_1.validator_store.voting_pubkeys();
|
||||
service_1
|
||||
@@ -588,13 +587,31 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
|
||||
let mut replaced = 0;
|
||||
let mut invalid = 0;
|
||||
|
||||
all_duties.into_iter().try_for_each::<_, Result<_, String>>(|remote_duties| {
|
||||
let duties: DutyAndState = remote_duties.try_into()?;
|
||||
// For each of the duties, attempt to insert them into our local store and build a
|
||||
// list of new or changed selections proofs for any aggregating validators.
|
||||
let validator_subscriptions = all_duties.into_iter().filter_map(|remote_duties| {
|
||||
// Convert the remote duties into our local representation.
|
||||
let duties: DutyAndProof = remote_duties
|
||||
.try_into()
|
||||
.map_err(|e| error!(
|
||||
log,
|
||||
"Unable to convert remote duties";
|
||||
"error" => e
|
||||
))
|
||||
.ok()?;
|
||||
|
||||
match service_2
|
||||
// Attempt to update our local store.
|
||||
let outcome = service_2
|
||||
.store
|
||||
.insert(epoch, duties.clone(), E::slots_per_epoch())
|
||||
{
|
||||
.insert(epoch, duties.clone(), E::slots_per_epoch(), &service_2.validator_store)
|
||||
.map_err(|e| error!(
|
||||
log,
|
||||
"Unable to store duties";
|
||||
"error" => e
|
||||
))
|
||||
.ok()?;
|
||||
|
||||
match &outcome {
|
||||
InsertOutcome::NewValidator => {
|
||||
debug!(
|
||||
log,
|
||||
@@ -603,16 +620,25 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
|
||||
"attestation_slot" => format!("{:?}", &duties.duty.attestation_slot),
|
||||
"validator" => format!("{:?}", &duties.duty.validator_pubkey)
|
||||
);
|
||||
new_validator += 1
|
||||
new_validator += 1;
|
||||
}
|
||||
InsertOutcome::NewEpoch => new_epoch += 1,
|
||||
InsertOutcome::Identical => identical += 1,
|
||||
InsertOutcome::Replaced => replaced += 1,
|
||||
InsertOutcome::Replaced { .. } => replaced += 1,
|
||||
InsertOutcome::Invalid => invalid += 1,
|
||||
};
|
||||
|
||||
Ok(())
|
||||
})?;
|
||||
if outcome.is_subscription_candidate() {
|
||||
Some(ValidatorSubscription {
|
||||
validator_index: duties.duty.validator_index?,
|
||||
attestation_committee_index: duties.duty.attestation_committee_index?,
|
||||
slot: duties.duty.attestation_slot?,
|
||||
is_aggregator: duties.selection_proof.is_some(),
|
||||
})
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}).collect::<Vec<_>>();
|
||||
|
||||
if invalid > 0 {
|
||||
error!(
|
||||
@@ -641,7 +667,51 @@ impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
|
||||
)
|
||||
}
|
||||
|
||||
Ok(())
|
||||
Ok(validator_subscriptions)
|
||||
})
|
||||
.and_then::<_, Box<dyn Future<Item = _, Error = _> + Send>>(move |validator_subscriptions| {
|
||||
let log = service_3.context.log.clone();
|
||||
let count = validator_subscriptions.len();
|
||||
|
||||
if count == 0 {
|
||||
debug!(
|
||||
log,
|
||||
"No new subscriptions required"
|
||||
);
|
||||
|
||||
Box::new(future::ok(()))
|
||||
} else {
|
||||
Box::new(service_3.beacon_node
|
||||
.http
|
||||
.validator()
|
||||
.subscribe(validator_subscriptions)
|
||||
.map_err(|e| format!("Failed to subscribe validators: {:?}", e))
|
||||
.map(move |status| {
|
||||
match status {
|
||||
PublishStatus::Valid => {
|
||||
debug!(
|
||||
log,
|
||||
"Successfully subscribed validators";
|
||||
"count" => count
|
||||
)
|
||||
},
|
||||
PublishStatus::Unknown => {
|
||||
error!(
|
||||
log,
|
||||
"Unknown response from subscription";
|
||||
)
|
||||
},
|
||||
PublishStatus::Invalid(e) => {
|
||||
error!(
|
||||
log,
|
||||
"Failed to subscribe validator";
|
||||
"error" => e
|
||||
)
|
||||
},
|
||||
};
|
||||
}))
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -224,6 +224,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
|
||||
validator_pubkey: &PublicKey,
|
||||
validator_index: u64,
|
||||
aggregate: Attestation<E>,
|
||||
selection_proof: SelectionProof,
|
||||
) -> Option<SignedAggregateAndProof<E>> {
|
||||
let validators = self.validators.read();
|
||||
let voting_keypair = validators.get(validator_pubkey)?.voting_keypair.as_ref()?;
|
||||
@@ -231,6 +232,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
|
||||
Some(SignedAggregateAndProof::from_aggregate(
|
||||
validator_index,
|
||||
aggregate,
|
||||
Some(selection_proof),
|
||||
&voting_keypair.sk,
|
||||
&self.fork()?,
|
||||
self.genesis_validators_root,
|
||||
|
||||
Reference in New Issue
Block a user