gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-23 06:44:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Squashed reset to unstable

Browse Source
This commit is contained in:
Daniel Knopik
2025-03-13 12:50:29 +01:00
committed by Daniel Knopik
parent b71b5f2231
commit f61f0b654c
416 changed files with 13195 additions and 38478 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
validator_client/validator_services/Cargo.toml
View File

@@ -6,8 +6,10 @@ authors = ["Sigma Prime <contact@sigmaprime.io>"]
[dependencies]
beacon_node_fallback = { workspace = true }
bls = { workspace = true }
bls = { workspace = true }
doppelganger_service = { workspace = true }
either = { workspace = true }
environment = { workspace = true }
eth2 = { workspace = true }
futures = { workspace = true }
graffiti_file = { workspace = true }
@@ -15,7 +17,6 @@ logging = { workspace = true }
parking_lot = { workspace = true }
safe_arith = { workspace = true }
slot_clock = { workspace = true }
task_executor = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
tree_hash = { workspace = true }

113
validator_client/validator_services/src/attestation_service.rs
View File

@@ -1,13 +1,13 @@
use crate::duties_service::{DutiesService, DutyAndProof};
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
use either::Either;
use environment::RuntimeContext;
use futures::future::join_all;
use logging::crit;
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use task_executor::TaskExecutor;
use tokio::time::{sleep, sleep_until, Duration, Instant};
use tracing::{debug, error, info, trace, warn};
use tree_hash::TreeHash;
@@ -16,33 +16,33 @@ use validator_store::{Error as ValidatorStoreError, ValidatorStore};
/// Builds an `AttestationService`.
#[derive(Default)]
pub struct AttestationServiceBuilder<S: ValidatorStore, T: SlotClock + 'static> {
duties_service: Option<Arc<DutiesService<S, T>>>,
validator_store: Option<Arc<S>>,
pub struct AttestationServiceBuilder<T: SlotClock + 'static, E: EthSpec> {
duties_service: Option<Arc<DutiesService<T, E>>>,
validator_store: Option<Arc<ValidatorStore<T, E>>>,
slot_clock: Option<T>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T>>>,
executor: Option<TaskExecutor>,
chain_spec: Option<Arc<ChainSpec>>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
context: Option<RuntimeContext<E>>,
disable: bool,
}
impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuilder<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> AttestationServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
duties_service: None,
validator_store: None,
slot_clock: None,
beacon_nodes: None,
executor: None,
chain_spec: None,
context: None,
disable: false,
}
}
pub fn duties_service(mut self, service: Arc<DutiesService<S, T>>) -> Self {
pub fn duties_service(mut self, service: Arc<DutiesService<T, E>>) -> Self {
self.duties_service = Some(service);
self
}
pub fn validator_store(mut self, store: Arc<S>) -> Self {
pub fn validator_store(mut self, store: Arc<ValidatorStore<T, E>>) -> Self {
self.validator_store = Some(store);
self
}
@@ -52,22 +52,22 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
self
}
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
self.beacon_nodes = Some(beacon_nodes);
self
}
pub fn executor(mut self, executor: TaskExecutor) -> Self {
self.executor = Some(executor);
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
pub fn chain_spec(mut self, chain_spec: Arc<ChainSpec>) -> Self {
self.chain_spec = Some(chain_spec);
pub fn disable(mut self, disable: bool) -> Self {
self.disable = disable;
self
}
pub fn build(self) -> Result<AttestationService<S, T>, String> {
pub fn build(self) -> Result<AttestationService<T, E>, String> {
Ok(AttestationService {
inner: Arc::new(Inner {
duties_service: self
@@ -82,25 +82,23 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationServiceBuil
beacon_nodes: self
.beacon_nodes
.ok_or("Cannot build AttestationService without beacon_nodes")?,
executor: self
.executor
.ok_or("Cannot build AttestationService without executor")?,
chain_spec: self
.chain_spec
.ok_or("Cannot build AttestationService without chain_spec")?,
context: self
.context
.ok_or("Cannot build AttestationService without runtime_context")?,
disable: self.disable,
}),
})
}
}
/// Helper to minimise `Arc` usage.
pub struct Inner<S, T> {
duties_service: Arc<DutiesService<S, T>>,
validator_store: Arc<S>,
pub struct Inner<T, E: EthSpec> {
duties_service: Arc<DutiesService<T, E>>,
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T>>,
executor: TaskExecutor,
chain_spec: Arc<ChainSpec>,
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
context: RuntimeContext<E>,
disable: bool,
}
/// Attempts to produce attestations for all known validators 1/3rd of the way through each slot.
@@ -108,11 +106,11 @@ pub struct Inner<S, T> {
/// 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<S, T> {
inner: Arc<Inner<S, T>>,
pub struct AttestationService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<S, T> Clone for AttestationService<S, T> {
impl<T, E: EthSpec> Clone for AttestationService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -120,17 +118,22 @@ impl<S, T> Clone for AttestationService<S, T> {
}
}
impl<S, T> Deref for AttestationService<S, T> {
type Target = Inner<S, T>;
impl<T, E: EthSpec> Deref for AttestationService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S, T> {
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<(), String> {
if self.disable {
info!("Attestation service disabled");
return Ok(());
}
let slot_duration = Duration::from_secs(spec.seconds_per_slot);
let duration_to_next_slot = self
.slot_clock
@@ -142,7 +145,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
"Attestation production service started"
);
let executor = self.executor.clone();
let executor = self.context.executor.clone();
let interval_fut = async move {
loop {
@@ -202,7 +205,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
.into_iter()
.for_each(|(committee_index, validator_duties)| {
// Spawn a separate task for each attestation.
self.inner.executor.spawn_ignoring_error(
self.inner.context.executor.spawn_ignoring_error(
self.clone().publish_attestations_and_aggregates(
slot,
committee_index,
@@ -329,7 +332,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
.slot_clock
.now()
.ok_or("Unable to determine current slot from clock")?
.epoch(S::E::slots_per_epoch());
.epoch(E::slots_per_epoch());
let attestation_data = self
.beacon_nodes
@@ -354,10 +357,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
let attestation_data = attestation_data_ref;
// Ensure that the attestation matches the duties.
if !duty.match_attestation_data::<S::E>(attestation_data, &self.chain_spec) {
if !duty.match_attestation_data::<E>(attestation_data, &self.context.eth2_config.spec) {
crit!(
validator = ?duty.pubkey,
duty_slot = ?duty.slot,
duty_slot = %duty.slot,
attestation_slot = %attestation_data.slot,
duty_index = duty.committee_index,
attestation_index = attestation_data.index,
@@ -366,14 +369,14 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
return None;
}
let mut attestation = match Attestation::empty_for_signing(
let mut attestation = match Attestation::<E>::empty_for_signing(
duty.committee_index,
duty.committee_length as usize,
attestation_data.slot,
attestation_data.beacon_block_root,
attestation_data.source,
attestation_data.target,
&self.chain_spec,
&self.context.eth2_config.spec,
) {
Ok(attestation) => attestation,
Err(err) => {
@@ -436,8 +439,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
return Ok(None);
}
let fork_name = self
.chain_spec
.fork_name_at_slot::<S::E>(attestation_data.slot);
.context
.eth2_config
.spec
.fork_name_at_slot::<E>(attestation_data.slot);
// Post the attestations to the BN.
match self
@@ -462,7 +467,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
committee_index = attestation_data.index,
slot = slot.as_u64(),
"type" = "unaggregated",
"Unable to convert to SingleAttestation",
"Unable to convert to SingleAttestation"
);
None
}
@@ -471,7 +476,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
.collect::<Vec<_>>();
beacon_node
.post_beacon_pool_attestations_v2::<S::E>(
.post_beacon_pool_attestations_v2::<E>(
Either::Right(single_attestations),
fork_name,
)
@@ -533,8 +538,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
}
let fork_name = self
.chain_spec
.fork_name_at_slot::<S::E>(attestation_data.slot);
.context
.eth2_config
.spec
.fork_name_at_slot::<E>(attestation_data.slot);
let aggregated_attestation = &self
.beacon_nodes
@@ -578,7 +585,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
let duty = &duty_and_proof.duty;
let selection_proof = duty_and_proof.selection_proof.as_ref()?;
if !duty.match_attestation_data::<S::E>(attestation_data, &self.chain_spec) {
if !duty.match_attestation_data::<E>(attestation_data, &self.context.eth2_config.spec) {
crit!("Inconsistent validator duties during signing");
return None;
}
@@ -682,11 +689,11 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
/// Start the task at `pruning_instant` to avoid interference with other tasks.
fn spawn_slashing_protection_pruning_task(&self, slot: Slot, pruning_instant: Instant) {
let attestation_service = self.clone();
let executor = self.inner.executor.clone();
let current_epoch = slot.epoch(S::E::slots_per_epoch());
let executor = self.inner.context.executor.clone();
let current_epoch = slot.epoch(E::slots_per_epoch());
// Wait for `pruning_instant` in a regular task, and then switch to a blocking one.
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
sleep_until(pruning_instant).await;

134
validator_client/validator_services/src/block_service.rs
View File

@@ -1,5 +1,6 @@
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback, Error as FallbackError, Errors};
use bls::SignatureBytes;
use environment::RuntimeContext;
use eth2::types::{FullBlockContents, PublishBlockRequest};
use eth2::{BeaconNodeHttpClient, StatusCode};
use graffiti_file::{determine_graffiti, GraffitiFile};
@@ -10,12 +11,11 @@ use std::future::Future;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use task_executor::TaskExecutor;
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace, warn};
use types::{
BlindedBeaconBlock, BlockType, ChainSpec, EthSpec, Graffiti, PublicKeyBytes,
SignedBlindedBeaconBlock, Slot,
BlindedBeaconBlock, BlockType, EthSpec, Graffiti, PublicKeyBytes, SignedBlindedBeaconBlock,
Slot,
};
use validator_store::{Error as ValidatorStoreError, ValidatorStore};
@@ -45,32 +45,30 @@ impl From<Errors<BlockError>> for BlockError {
/// Builds a `BlockService`.
#[derive(Default)]
pub struct BlockServiceBuilder<S, T> {
validator_store: Option<Arc<S>>,
pub struct BlockServiceBuilder<T, E: EthSpec> {
validator_store: Option<Arc<ValidatorStore<T, E>>>,
slot_clock: Option<Arc<T>>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T>>>,
proposer_nodes: Option<Arc<BeaconNodeFallback<T>>>,
executor: Option<TaskExecutor>,
chain_spec: Option<Arc<ChainSpec>>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
context: Option<RuntimeContext<E>>,
graffiti: Option<Graffiti>,
graffiti_file: Option<GraffitiFile>,
}
impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
validator_store: None,
slot_clock: None,
beacon_nodes: None,
proposer_nodes: None,
executor: None,
chain_spec: None,
context: None,
graffiti: None,
graffiti_file: None,
}
}
pub fn validator_store(mut self, store: Arc<S>) -> Self {
pub fn validator_store(mut self, store: Arc<ValidatorStore<T, E>>) -> Self {
self.validator_store = Some(store);
self
}
@@ -80,23 +78,18 @@ impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
self
}
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
self.beacon_nodes = Some(beacon_nodes);
self
}
pub fn proposer_nodes(mut self, proposer_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
pub fn proposer_nodes(mut self, proposer_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
self.proposer_nodes = Some(proposer_nodes);
self
}
pub fn executor(mut self, executor: TaskExecutor) -> Self {
self.executor = Some(executor);
self
}
pub fn chain_spec(mut self, chain_spec: Arc<ChainSpec>) -> Self {
self.chain_spec = Some(chain_spec);
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
@@ -110,7 +103,7 @@ impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
self
}
pub fn build(self) -> Result<BlockService<S, T>, String> {
pub fn build(self) -> Result<BlockService<T, E>, String> {
Ok(BlockService {
inner: Arc::new(Inner {
validator_store: self
@@ -122,12 +115,9 @@ impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
beacon_nodes: self
.beacon_nodes
.ok_or("Cannot build BlockService without beacon_node")?,
executor: self
.executor
.ok_or("Cannot build BlockService without executor")?,
chain_spec: self
.chain_spec
.ok_or("Cannot build BlockService without chain_spec")?,
context: self
.context
.ok_or("Cannot build BlockService without runtime_context")?,
proposer_nodes: self.proposer_nodes,
graffiti: self.graffiti,
graffiti_file: self.graffiti_file,
@@ -138,12 +128,12 @@ impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
// Combines a set of non-block-proposing `beacon_nodes` and only-block-proposing
// `proposer_nodes`.
pub struct ProposerFallback<T> {
beacon_nodes: Arc<BeaconNodeFallback<T>>,
proposer_nodes: Option<Arc<BeaconNodeFallback<T>>>,
pub struct ProposerFallback<T, E: EthSpec> {
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
}
impl<T: SlotClock> ProposerFallback<T> {
impl<T: SlotClock, E: EthSpec> ProposerFallback<T, E> {
// Try `func` on `self.proposer_nodes` first. If that doesn't work, try `self.beacon_nodes`.
pub async fn request_proposers_first<F, Err, R>(&self, func: F) -> Result<(), Errors<Err>>
where
@@ -188,23 +178,22 @@ impl<T: SlotClock> ProposerFallback<T> {
}
/// Helper to minimise `Arc` usage.
pub struct Inner<S, T> {
validator_store: Arc<S>,
pub struct Inner<T, E: EthSpec> {
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: Arc<T>,
pub beacon_nodes: Arc<BeaconNodeFallback<T>>,
pub proposer_nodes: Option<Arc<BeaconNodeFallback<T>>>,
executor: TaskExecutor,
chain_spec: Arc<ChainSpec>,
pub beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
pub proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
context: RuntimeContext<E>,
graffiti: Option<Graffiti>,
graffiti_file: Option<GraffitiFile>,
}
/// Attempts to produce attestations for any block producer(s) at the start of the epoch.
pub struct BlockService<S, T> {
inner: Arc<Inner<S, T>>,
pub struct BlockService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<S, T> Clone for BlockService<S, T> {
impl<T, E: EthSpec> Clone for BlockService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -212,8 +201,8 @@ impl<S, T> Clone for BlockService<S, T> {
}
}
impl<S, T> Deref for BlockService<S, T> {
type Target = Inner<S, T>;
impl<T, E: EthSpec> Deref for BlockService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
@@ -226,14 +215,14 @@ pub struct BlockServiceNotification {
pub block_proposers: Vec<PublicKeyBytes>,
}
impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
pub fn start_update_service(
self,
mut notification_rx: mpsc::Receiver<BlockServiceNotification>,
) -> Result<(), String> {
info!("Block production service started");
let executor = self.inner.executor.clone();
let executor = self.inner.context.executor.clone();
executor.spawn(
async move {
@@ -269,7 +258,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
return Ok(());
}
if slot == self.chain_spec.genesis_slot {
if slot == self.context.eth2_config.spec.genesis_slot {
debug!(
proposers = format!("{:?}", notification.block_proposers),
"Not producing block at genesis slot"
@@ -298,7 +287,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
for validator_pubkey in proposers {
let builder_boost_factor = self.get_builder_boost_factor(&validator_pubkey);
let service = self.clone();
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
let result = service
.publish_block(slot, validator_pubkey, builder_boost_factor)
@@ -325,35 +314,30 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
#[allow(clippy::too_many_arguments)]
async fn sign_and_publish_block(
&self,
proposer_fallback: ProposerFallback<T>,
proposer_fallback: ProposerFallback<T, E>,
slot: Slot,
graffiti: Option<Graffiti>,
validator_pubkey: &PublicKeyBytes,
unsigned_block: UnsignedBlock<S::E>,
unsigned_block: UnsignedBlock<E>,
) -> Result<(), BlockError> {
let signing_timer = validator_metrics::start_timer(&validator_metrics::BLOCK_SIGNING_TIMES);
let (block, maybe_blobs) = match unsigned_block {
let res = match unsigned_block {
UnsignedBlock::Full(block_contents) => {
let (block, maybe_blobs) = block_contents.deconstruct();
(block.into(), maybe_blobs)
self.validator_store
.sign_block(*validator_pubkey, block, slot)
.await
.map(|b| SignedBlock::Full(PublishBlockRequest::new(Arc::new(b), maybe_blobs)))
}
UnsignedBlock::Blinded(block) => (block.into(), None),
UnsignedBlock::Blinded(block) => self
.validator_store
.sign_block(*validator_pubkey, block, slot)
.await
.map(Arc::new)
.map(SignedBlock::Blinded),
};
let res = self
.validator_store
.sign_block(*validator_pubkey, block, slot)
.await
.map(|block| match block {
validator_store::SignedBlock::Full(block) => {
SignedBlock::Full(PublishBlockRequest::new(Arc::new(block), maybe_blobs))
}
validator_store::SignedBlock::Blinded(block) => {
SignedBlock::Blinded(Arc::new(block))
}
});
let signed_block = match res {
Ok(block) => block,
Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
@@ -420,7 +404,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
let randao_reveal = match self
.validator_store
.randao_reveal(validator_pubkey, slot.epoch(S::E::slots_per_epoch()))
.randao_reveal(validator_pubkey, slot.epoch(E::slots_per_epoch()))
.await
{
Ok(signature) => signature.into(),
@@ -503,7 +487,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
async fn publish_signed_block_contents(
&self,
signed_block: &SignedBlock<S::E>,
signed_block: &SignedBlock<E>,
beacon_node: BeaconNodeHttpClient,
) -> Result<(), BlockError> {
let slot = signed_block.slot();
@@ -539,9 +523,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
graffiti: Option<Graffiti>,
proposer_index: Option<u64>,
builder_boost_factor: Option<u64>,
) -> Result<UnsignedBlock<S::E>, BlockError> {
) -> Result<UnsignedBlock<E>, BlockError> {
let (block_response, _) = beacon_node
.get_validator_blocks_v3::<S::E>(
.get_validator_blocks_v3::<E>(
slot,
randao_reveal_ref,
graffiti.as_ref(),
@@ -579,9 +563,15 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
// Apply per validator configuration first.
let validator_builder_boost_factor = self
.validator_store
.determine_builder_boost_factor(validator_pubkey);
.determine_validator_builder_boost_factor(validator_pubkey);
if let Some(builder_boost_factor) = validator_builder_boost_factor {
// Fallback to process-wide configuration if needed.
let maybe_builder_boost_factor = validator_builder_boost_factor.or_else(|| {
self.validator_store
.determine_default_builder_boost_factor()
});
if let Some(builder_boost_factor) = maybe_builder_boost_factor {
// if builder boost factor is set to 100 it should be treated
// as None to prevent unnecessary calculations that could
// lead to loss of information.

250
validator_client/validator_services/src/duties_service.rs
View File

@@ -10,6 +10,8 @@ use crate::block_service::BlockServiceNotification;
use crate::sync::poll_sync_committee_duties;
use crate::sync::SyncDutiesMap;
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
use doppelganger_service::DoppelgangerStatus;
use environment::RuntimeContext;
use eth2::types::{
AttesterData, BeaconCommitteeSubscription, DutiesResponse, ProposerData, StateId, ValidatorId,
};
@@ -22,12 +24,11 @@ use std::collections::{hash_map, BTreeMap, HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use task_executor::TaskExecutor;
use tokio::{sync::mpsc::Sender, time::sleep};
use tracing::{debug, error, info, warn};
use types::{ChainSpec, Epoch, EthSpec, Hash256, PublicKeyBytes, SelectionProof, Slot};
use validator_metrics::{get_int_gauge, set_int_gauge, ATTESTATION_DUTY};
use validator_store::{DoppelgangerStatus, Error as ValidatorStoreError, ValidatorStore};
use validator_store::{Error as ValidatorStoreError, ValidatorStore};
/// Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch.
const HISTORICAL_DUTIES_EPOCHS: u64 = 2;
@@ -86,16 +87,16 @@ const _: () = assert!(ATTESTATION_SUBSCRIPTION_OFFSETS[0] > MIN_ATTESTATION_SUBS
// The info in the enum variants is displayed in logging, clippy thinks it's dead code.
#[derive(Debug)]
pub enum Error<T> {
pub enum Error {
UnableToReadSlotClock,
FailedToDownloadAttesters(#[allow(dead_code)] String),
FailedToProduceSelectionProof(#[allow(dead_code)] ValidatorStoreError<T>),
FailedToProduceSelectionProof(#[allow(dead_code)] ValidatorStoreError),
InvalidModulo(#[allow(dead_code)] ArithError),
Arith(#[allow(dead_code)] ArithError),
SyncDutiesNotFound(#[allow(dead_code)] u64),
}
impl<T> From<ArithError> for Error<T> {
impl From<ArithError> for Error {
fn from(e: ArithError) -> Self {
Self::Arith(e)
}
@@ -124,11 +125,11 @@ pub struct SubscriptionSlots {
/// Create a selection proof for `duty`.
///
/// Return `Ok(None)` if the attesting validator is not an aggregator.
async fn make_selection_proof<S: ValidatorStore + 'static>(
async fn make_selection_proof<T: SlotClock + 'static, E: EthSpec>(
duty: &AttesterData,
validator_store: &S,
validator_store: &ValidatorStore<T, E>,
spec: &ChainSpec,
) -> Result<Option<SelectionProof>, Error<S::Error>> {
) -> Result<Option<SelectionProof>, Error> {
let selection_proof = validator_store
.produce_selection_proof(duty.pubkey, duty.slot)
.await
@@ -204,133 +205,35 @@ type DependentRoot = Hash256;
type AttesterMap = HashMap<PublicKeyBytes, HashMap<Epoch, (DependentRoot, DutyAndProof)>>;
type ProposerMap = HashMap<Epoch, (DependentRoot, Vec<ProposerData>)>;
pub struct DutiesServiceBuilder<S, T> {
/// Provides the canonical list of locally-managed validators.
validator_store: Option<Arc<S>>,
/// Tracks the current slot.
slot_clock: Option<T>,
/// Provides HTTP access to remote beacon nodes.
beacon_nodes: Option<Arc<BeaconNodeFallback<T>>>,
/// The runtime for spawning tasks.
executor: Option<TaskExecutor>,
/// The current chain spec.
spec: Option<Arc<ChainSpec>>,
//// Whether we permit large validator counts in the metrics.
enable_high_validator_count_metrics: bool,
/// If this validator is running in distributed mode.
distributed: bool,
}
impl<S, T> Default for DutiesServiceBuilder<S, T> {
fn default() -> Self {
Self::new()
}
}
impl<S, T> DutiesServiceBuilder<S, T> {
pub fn new() -> Self {
Self {
validator_store: None,
slot_clock: None,
beacon_nodes: None,
executor: None,
spec: None,
enable_high_validator_count_metrics: false,
distributed: false,
}
}
pub fn validator_store(mut self, validator_store: Arc<S>) -> Self {
self.validator_store = Some(validator_store);
self
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(slot_clock);
self
}
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
self.beacon_nodes = Some(beacon_nodes);
self
}
pub fn executor(mut self, executor: TaskExecutor) -> Self {
self.executor = Some(executor);
self
}
pub fn spec(mut self, spec: Arc<ChainSpec>) -> Self {
self.spec = Some(spec);
self
}
pub fn enable_high_validator_count_metrics(
mut self,
enable_high_validator_count_metrics: bool,
) -> Self {
self.enable_high_validator_count_metrics = enable_high_validator_count_metrics;
self
}
pub fn distributed(mut self, distributed: bool) -> Self {
self.distributed = distributed;
self
}
pub fn build(self) -> Result<DutiesService<S, T>, String> {
Ok(DutiesService {
attesters: Default::default(),
proposers: Default::default(),
sync_duties: SyncDutiesMap::new(self.distributed),
validator_store: self
.validator_store
.ok_or("Cannot build DutiesService without validator_store")?,
unknown_validator_next_poll_slots: Default::default(),
slot_clock: self
.slot_clock
.ok_or("Cannot build DutiesService without slot_clock")?,
beacon_nodes: self
.beacon_nodes
.ok_or("Cannot build DutiesService without beacon_nodes")?,
executor: self
.executor
.ok_or("Cannot build DutiesService without executor")?,
spec: self.spec.ok_or("Cannot build DutiesService without spec")?,
enable_high_validator_count_metrics: self.enable_high_validator_count_metrics,
distributed: self.distributed,
})
}
}
/// See the module-level documentation.
pub struct DutiesService<S, T> {
pub struct DutiesService<T, E: EthSpec> {
/// Maps a validator public key to their duties for each epoch.
pub attesters: RwLock<AttesterMap>,
/// Maps an epoch to all *local* proposers in this epoch. Notably, this does not contain
/// proposals for any validators which are not registered locally.
pub proposers: RwLock<ProposerMap>,
/// Map from validator index to sync committee duties.
pub sync_duties: SyncDutiesMap,
pub sync_duties: SyncDutiesMap<E>,
/// Provides the canonical list of locally-managed validators.
pub validator_store: Arc<S>,
pub validator_store: Arc<ValidatorStore<T, E>>,
/// Maps unknown validator pubkeys to the next slot time when a poll should be conducted again.
pub unknown_validator_next_poll_slots: RwLock<HashMap<PublicKeyBytes, Slot>>,
/// Tracks the current slot.
pub slot_clock: T,
/// Provides HTTP access to remote beacon nodes.
pub beacon_nodes: Arc<BeaconNodeFallback<T>>,
pub beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
/// The runtime for spawning tasks.
pub executor: TaskExecutor,
pub context: RuntimeContext<E>,
/// The current chain spec.
pub spec: Arc<ChainSpec>,
//// Whether we permit large validator counts in the metrics.
pub enable_high_validator_count_metrics: bool,
/// If this validator is running in distributed mode.
pub distributed: bool,
pub disable_attesting: bool,
}
impl<S: ValidatorStore, T: SlotClock + 'static> DutiesService<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
/// Returns the total number of validators known to the duties service.
pub fn total_validator_count(&self) -> usize {
self.validator_store.num_voting_validators()
@@ -380,8 +283,8 @@ impl<S: ValidatorStore, T: SlotClock + 'static> DutiesService<S, T> {
///
/// It is possible that multiple validators have an identical proposal slot, however that is
/// likely the result of heavy forking (lol) or inconsistent beacon node connections.
pub fn block_proposers<E: EthSpec>(&self, slot: Slot) -> HashSet<PublicKeyBytes> {
let epoch = slot.epoch(S::E::slots_per_epoch());
pub fn block_proposers(&self, slot: Slot) -> HashSet<PublicKeyBytes> {
let epoch = slot.epoch(E::slots_per_epoch());
// Only collect validators that are considered safe in terms of doppelganger protection.
let signing_pubkeys: HashSet<_> = self
@@ -406,7 +309,7 @@ impl<S: ValidatorStore, T: SlotClock + 'static> DutiesService<S, T> {
/// Returns all `ValidatorDuty` for the given `slot`.
pub fn attesters(&self, slot: Slot) -> Vec<DutyAndProof> {
let epoch = slot.epoch(S::E::slots_per_epoch());
let epoch = slot.epoch(E::slots_per_epoch());
// Only collect validators that are considered safe in terms of doppelganger protection.
let signing_pubkeys: HashSet<_> = self
@@ -444,15 +347,15 @@ impl<S: ValidatorStore, T: SlotClock + 'static> DutiesService<S, T> {
/// process every slot, which has the chance of creating a theoretically unlimited backlog of tasks.
/// It was a conscious decision to choose to drop tasks on an overloaded/latent system rather than
/// overload it even more.
pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>(
core_duties_service: Arc<DutiesService<S, T>>,
pub fn start_update_service<T: SlotClock + 'static, E: EthSpec>(
core_duties_service: Arc<DutiesService<T, E>>,
mut block_service_tx: Sender<BlockServiceNotification>,
) {
/*
* Spawn the task which updates the map of pubkey to validator index.
*/
let duties_service = core_duties_service.clone();
core_duties_service.executor.spawn(
core_duties_service.context.executor.spawn(
async move {
loop {
// Run this poll before the wait, this should hopefully download all the indices
@@ -475,7 +378,7 @@ pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>
* Spawn the task which keeps track of local block proposal duties.
*/
let duties_service = core_duties_service.clone();
core_duties_service.executor.spawn(
core_duties_service.context.executor.spawn(
async move {
loop {
if let Some(duration) = duties_service.slot_clock.duration_to_next_slot() {
@@ -499,11 +402,16 @@ pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>
"duties_service_proposers",
);
// Skip starting attestation duties or sync committee services.
if core_duties_service.disable_attesting {
return;
}
/*
* Spawn the task which keeps track of local attestation duties.
*/
let duties_service = core_duties_service.clone();
core_duties_service.executor.spawn(
core_duties_service.context.executor.spawn(
async move {
loop {
if let Some(duration) = duties_service.slot_clock.duration_to_next_slot() {
@@ -528,7 +436,7 @@ pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>
// Spawn the task which keeps track of local sync committee duties.
let duties_service = core_duties_service.clone();
core_duties_service.executor.spawn(
core_duties_service.context.executor.spawn(
async move {
loop {
if let Err(e) = poll_sync_committee_duties(&duties_service).await {
@@ -558,8 +466,8 @@ pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>
/// Iterate through all the voting pubkeys in the `ValidatorStore` and attempt to learn any unknown
/// validator indices.
async fn poll_validator_indices<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &DutiesService<S, T>,
async fn poll_validator_indices<T: SlotClock + 'static, E: EthSpec>(
duties_service: &DutiesService<T, E>,
) {
let _timer = validator_metrics::start_timer_vec(
&validator_metrics::DUTIES_SERVICE_TIMES,
@@ -578,14 +486,16 @@ async fn poll_validator_indices<S: ValidatorStore, T: SlotClock + 'static>(
// This is on its own line to avoid some weirdness with locks and if statements.
let is_known = duties_service
.validator_store
.validator_index(&pubkey)
.initialized_validators()
.read()
.get_index(&pubkey)
.is_some();
if !is_known {
let current_slot_opt = duties_service.slot_clock.now();
if let Some(current_slot) = current_slot_opt {
let is_first_slot_of_epoch = current_slot % S::E::slots_per_epoch() == 0;
let is_first_slot_of_epoch = current_slot % E::slots_per_epoch() == 0;
// Query an unknown validator later if it was queried within the last epoch, or if
// the current slot is the first slot of an epoch.
@@ -636,7 +546,9 @@ async fn poll_validator_indices<S: ValidatorStore, T: SlotClock + 'static>(
);
duties_service
.validator_store
.set_validator_index(&pubkey, response.data.index);
.initialized_validators()
.write()
.set_index(&pubkey, response.data.index);
duties_service
.unknown_validator_next_poll_slots
@@ -647,7 +559,7 @@ async fn poll_validator_indices<S: ValidatorStore, T: SlotClock + 'static>(
// the beacon chain.
Ok(None) => {
if let Some(current_slot) = current_slot_opt {
let next_poll_slot = current_slot.saturating_add(S::E::slots_per_epoch());
let next_poll_slot = current_slot.saturating_add(E::slots_per_epoch());
duties_service
.unknown_validator_next_poll_slots
.write()
@@ -678,9 +590,9 @@ async fn poll_validator_indices<S: ValidatorStore, T: SlotClock + 'static>(
/// 2. As above, but for the next-epoch.
/// 3. Push out any attestation subnet subscriptions to the BN.
/// 4. Prune old entries from `duties_service.attesters`.
async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
) -> Result<(), Error<S::Error>> {
async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
) -> Result<(), Error> {
let current_epoch_timer = validator_metrics::start_timer_vec(
&validator_metrics::DUTIES_SERVICE_TIMES,
&[validator_metrics::UPDATE_ATTESTERS_CURRENT_EPOCH],
@@ -690,7 +602,7 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
.slot_clock
.now()
.ok_or(Error::UnableToReadSlotClock)?;
let current_epoch = current_slot.epoch(S::E::slots_per_epoch());
let current_epoch = current_slot.epoch(E::slots_per_epoch());
let next_epoch = current_epoch + 1;
// Collect *all* pubkeys, even those undergoing doppelganger protection.
@@ -704,8 +616,10 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
let local_indices = {
let mut local_indices = Vec::with_capacity(local_pubkeys.len());
let vals_ref = duties_service.validator_store.initialized_validators();
let vals = vals_ref.read();
for &pubkey in &local_pubkeys {
if let Some(validator_index) = duties_service.validator_store.validator_index(&pubkey) {
if let Some(validator_index) = vals.get_index(&pubkey) {
local_indices.push(validator_index)
}
}
@@ -729,7 +643,7 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
)
}
update_per_validator_duty_metrics(duties_service, current_epoch, current_slot);
update_per_validator_duty_metrics::<T, E>(duties_service, current_epoch, current_slot);
drop(current_epoch_timer);
let next_epoch_timer = validator_metrics::start_timer_vec(
@@ -750,7 +664,7 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
)
}
update_per_validator_duty_metrics(duties_service, next_epoch, current_slot);
update_per_validator_duty_metrics::<T, E>(duties_service, next_epoch, current_slot);
drop(next_epoch_timer);
let subscriptions_timer = validator_metrics::start_timer_vec(
@@ -771,7 +685,7 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
* std::cmp::max(
1,
local_pubkeys.len() * ATTESTATION_SUBSCRIPTION_OFFSETS.len()
/ S::E::slots_per_epoch() as usize,
/ E::slots_per_epoch() as usize,
)
/ overallocation_denominator;
let mut subscriptions = Vec::with_capacity(num_expected_subscriptions);
@@ -867,12 +781,12 @@ async fn poll_beacon_attesters<S: ValidatorStore + 'static, T: SlotClock + 'stat
/// For the given `local_indices` and `local_pubkeys`, download the duties for the given `epoch` and
/// store them in `duties_service.attesters`.
async fn poll_beacon_attesters_for_epoch<S: ValidatorStore + 'static, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
async fn poll_beacon_attesters_for_epoch<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
epoch: Epoch,
local_indices: &[u64],
local_pubkeys: &HashSet<PublicKeyBytes>,
) -> Result<(), Error<S::Error>> {
) -> Result<(), Error> {
// No need to bother the BN if we don't have any validators.
if local_indices.is_empty() {
debug!(
@@ -910,10 +824,10 @@ async fn poll_beacon_attesters_for_epoch<S: ValidatorStore + 'static, T: SlotClo
local_pubkeys
.iter()
.filter(|pubkey| {
attesters.get(pubkey).map_or(true, |duties| {
attesters.get(pubkey).is_none_or(|duties| {
duties
.get(&epoch)
.map_or(true, |(prior, _)| *prior != dependent_root)
.is_none_or(|(prior, _)| *prior != dependent_root)
})
})
.collect::<Vec<_>>()
@@ -1016,7 +930,7 @@ async fn poll_beacon_attesters_for_epoch<S: ValidatorStore + 'static, T: SlotClo
// Spawn the background task to compute selection proofs.
let subservice = duties_service.clone();
duties_service.executor.spawn(
duties_service.context.executor.spawn(
async move {
fill_in_selection_proofs(subservice, new_duties, dependent_root).await;
},
@@ -1027,8 +941,8 @@ async fn poll_beacon_attesters_for_epoch<S: ValidatorStore + 'static, T: SlotClo
}
/// Get a filtered list of local validators for which we don't already know their duties for that epoch
fn get_uninitialized_validators<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
fn get_uninitialized_validators<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
epoch: &Epoch,
local_pubkeys: &HashSet<PublicKeyBytes>,
) -> Vec<u64> {
@@ -1038,14 +952,14 @@ fn get_uninitialized_validators<S: ValidatorStore, T: SlotClock + 'static>(
.filter(|pubkey| {
attesters
.get(pubkey)
.map_or(true, |duties| !duties.contains_key(epoch))
.is_none_or(|duties| !duties.contains_key(epoch))
})
.filter_map(|pubkey| duties_service.validator_store.validator_index(pubkey))
.collect::<Vec<_>>()
}
fn update_per_validator_duty_metrics<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
fn update_per_validator_duty_metrics<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
epoch: Epoch,
current_slot: Slot,
) {
@@ -1060,14 +974,14 @@ fn update_per_validator_duty_metrics<S: ValidatorStore, T: SlotClock + 'static>(
get_int_gauge(&ATTESTATION_DUTY, &[&validator_index.to_string()])
{
let existing_slot = Slot::new(existing_slot_gauge.get() as u64);
let existing_epoch = existing_slot.epoch(S::E::slots_per_epoch());
let existing_epoch = existing_slot.epoch(E::slots_per_epoch());
// First condition ensures that we switch to the next epoch duty slot
// once the current epoch duty slot passes.
// Second condition is to ensure that next epoch duties don't override
// current epoch duties.
if existing_slot < current_slot
|| (duty_slot.epoch(S::E::slots_per_epoch()) <= existing_epoch
|| (duty_slot.epoch(E::slots_per_epoch()) <= existing_epoch
&& duty_slot > current_slot
&& duty_slot != existing_slot)
{
@@ -1085,11 +999,11 @@ fn update_per_validator_duty_metrics<S: ValidatorStore, T: SlotClock + 'static>(
}
}
async fn post_validator_duties_attester<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
async fn post_validator_duties_attester<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
epoch: Epoch,
validator_indices: &[u64],
) -> Result<DutiesResponse<Vec<AttesterData>>, Error<S::Error>> {
) -> Result<DutiesResponse<Vec<AttesterData>>, Error> {
duties_service
.beacon_nodes
.first_success(|beacon_node| async move {
@@ -1109,8 +1023,8 @@ async fn post_validator_duties_attester<S: ValidatorStore, T: SlotClock + 'stati
///
/// Duties are computed in batches each slot. If a re-org is detected then the process will
/// terminate early as it is assumed the selection proofs from `duties` are no longer relevant.
async fn fill_in_selection_proofs<S: ValidatorStore + 'static, T: SlotClock + 'static>(
duties_service: Arc<DutiesService<S, T>>,
async fn fill_in_selection_proofs<T: SlotClock + 'static, E: EthSpec>(
duties_service: Arc<DutiesService<T, E>>,
duties: Vec<AttesterData>,
dependent_root: Hash256,
) {
@@ -1161,7 +1075,7 @@ async fn fill_in_selection_proofs<S: ValidatorStore + 'static, T: SlotClock + 's
.then(|duty| async {
let opt_selection_proof = make_selection_proof(
&duty,
duties_service.validator_store.as_ref(),
&duties_service.validator_store,
&duties_service.spec,
)
.await?;
@@ -1200,7 +1114,7 @@ async fn fill_in_selection_proofs<S: ValidatorStore + 'static, T: SlotClock + 's
};
let attester_map = attesters.entry(duty.pubkey).or_default();
let epoch = duty.slot.epoch(S::E::slots_per_epoch());
let epoch = duty.slot.epoch(E::slots_per_epoch());
match attester_map.entry(epoch) {
hash_map::Entry::Occupied(mut entry) => {
// No need to update duties for which no proof was computed.
@@ -1277,10 +1191,10 @@ async fn fill_in_selection_proofs<S: ValidatorStore + 'static, T: SlotClock + 's
/// through the slow path every time. I.e., the proposal will only happen after we've been able to
/// download and process the duties from the BN. This means it is very important to ensure this
/// function is as fast as possible.
async fn poll_beacon_proposers<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &DutiesService<S, T>,
async fn poll_beacon_proposers<T: SlotClock + 'static, E: EthSpec>(
duties_service: &DutiesService<T, E>,
block_service_tx: &mut Sender<BlockServiceNotification>,
) -> Result<(), Error<S::Error>> {
) -> Result<(), Error> {
let _timer = validator_metrics::start_timer_vec(
&validator_metrics::DUTIES_SERVICE_TIMES,
&[validator_metrics::UPDATE_PROPOSERS],
@@ -1290,17 +1204,17 @@ async fn poll_beacon_proposers<S: ValidatorStore, T: SlotClock + 'static>(
.slot_clock
.now()
.ok_or(Error::UnableToReadSlotClock)?;
let current_epoch = current_slot.epoch(S::E::slots_per_epoch());
let current_epoch = current_slot.epoch(E::slots_per_epoch());
// Notify the block proposal service for any proposals that we have in our cache.
//
// See the function-level documentation for more information.
let initial_block_proposers = duties_service.block_proposers::<S::E>(current_slot);
notify_block_production_service::<S>(
let initial_block_proposers = duties_service.block_proposers(current_slot);
notify_block_production_service(
current_slot,
&initial_block_proposers,
block_service_tx,
duties_service.validator_store.as_ref(),
&duties_service.validator_store,
)
.await;
@@ -1372,7 +1286,7 @@ async fn poll_beacon_proposers<S: ValidatorStore, T: SlotClock + 'static>(
// Then, compute the difference between these two sets to obtain a set of block proposers
// which were not included in the initial notification to the `BlockService`.
let additional_block_producers = duties_service
.block_proposers::<S::E>(current_slot)
.block_proposers(current_slot)
.difference(&initial_block_proposers)
.copied()
.collect::<HashSet<PublicKeyBytes>>();
@@ -1382,11 +1296,11 @@ async fn poll_beacon_proposers<S: ValidatorStore, T: SlotClock + 'static>(
//
// See the function-level documentation for more reasoning about this behaviour.
if !additional_block_producers.is_empty() {
notify_block_production_service::<S>(
notify_block_production_service(
current_slot,
&additional_block_producers,
block_service_tx,
duties_service.validator_store.as_ref(),
&duties_service.validator_store,
)
.await;
debug!(
@@ -1407,11 +1321,11 @@ async fn poll_beacon_proposers<S: ValidatorStore, T: SlotClock + 'static>(
}
/// Notify the block service if it should produce a block.
async fn notify_block_production_service<S: ValidatorStore>(
async fn notify_block_production_service<T: SlotClock + 'static, E: EthSpec>(
current_slot: Slot,
block_proposers: &HashSet<PublicKeyBytes>,
block_service_tx: &mut Sender<BlockServiceNotification>,
validator_store: &S,
validator_store: &ValidatorStore<T, E>,
) {
let non_doppelganger_proposers = block_proposers
.iter()

98
validator_client/validator_services/src/preparation_service.rs
View File

@@ -1,5 +1,7 @@
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
use bls::PublicKeyBytes;
use doppelganger_service::DoppelgangerStatus;
use environment::RuntimeContext;
use parking_lot::RwLock;
use slot_clock::SlotClock;
use std::collections::HashMap;
@@ -7,16 +9,13 @@ use std::hash::Hash;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use task_executor::TaskExecutor;
use tokio::time::{sleep, Duration};
use tracing::{debug, error, info, warn};
use types::{
Address, ChainSpec, EthSpec, ProposerPreparationData, SignedValidatorRegistrationData,
ValidatorRegistrationData,
};
use validator_store::{
DoppelgangerStatus, Error as ValidatorStoreError, ProposalData, ValidatorStore,
};
use validator_store::{Error as ValidatorStoreError, ProposalData, ValidatorStore};
/// Number of epochs before the Bellatrix hard fork to begin posting proposer preparations.
const PROPOSER_PREPARATION_LOOKAHEAD_EPOCHS: u64 = 2;
@@ -26,28 +25,28 @@ const EPOCHS_PER_VALIDATOR_REGISTRATION_SUBMISSION: u64 = 1;
/// Builds an `PreparationService`.
#[derive(Default)]
pub struct PreparationServiceBuilder<S: ValidatorStore, T: SlotClock + 'static> {
validator_store: Option<Arc<S>>,
pub struct PreparationServiceBuilder<T: SlotClock + 'static, E: EthSpec> {
validator_store: Option<Arc<ValidatorStore<T, E>>>,
slot_clock: Option<T>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T>>>,
executor: Option<TaskExecutor>,
beacon_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
context: Option<RuntimeContext<E>>,
builder_registration_timestamp_override: Option<u64>,
validator_registration_batch_size: Option<usize>,
}
impl<S: ValidatorStore, T: SlotClock + 'static> PreparationServiceBuilder<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> PreparationServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
validator_store: None,
slot_clock: None,
beacon_nodes: None,
executor: None,
context: None,
builder_registration_timestamp_override: None,
validator_registration_batch_size: None,
}
}
pub fn validator_store(mut self, store: Arc<S>) -> Self {
pub fn validator_store(mut self, store: Arc<ValidatorStore<T, E>>) -> Self {
self.validator_store = Some(store);
self
}
@@ -57,13 +56,13 @@ impl<S: ValidatorStore, T: SlotClock + 'static> PreparationServiceBuilder<S, T>
self
}
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T>>) -> Self {
pub fn beacon_nodes(mut self, beacon_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
self.beacon_nodes = Some(beacon_nodes);
self
}
pub fn executor(mut self, executor: TaskExecutor) -> Self {
self.executor = Some(executor);
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
@@ -83,7 +82,7 @@ impl<S: ValidatorStore, T: SlotClock + 'static> PreparationServiceBuilder<S, T>
self
}
pub fn build(self) -> Result<PreparationService<S, T>, String> {
pub fn build(self) -> Result<PreparationService<T, E>, String> {
Ok(PreparationService {
inner: Arc::new(Inner {
validator_store: self
@@ -95,9 +94,9 @@ impl<S: ValidatorStore, T: SlotClock + 'static> PreparationServiceBuilder<S, T>
beacon_nodes: self
.beacon_nodes
.ok_or("Cannot build PreparationService without beacon_nodes")?,
executor: self
.executor
.ok_or("Cannot build PreparationService without executor")?,
context: self
.context
.ok_or("Cannot build PreparationService without runtime_context")?,
builder_registration_timestamp_override: self
.builder_registration_timestamp_override,
validator_registration_batch_size: self.validator_registration_batch_size.ok_or(
@@ -110,11 +109,11 @@ impl<S: ValidatorStore, T: SlotClock + 'static> PreparationServiceBuilder<S, T>
}
/// Helper to minimise `Arc` usage.
pub struct Inner<S, T> {
validator_store: Arc<S>,
pub struct Inner<T, E: EthSpec> {
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T>>,
executor: TaskExecutor,
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
context: RuntimeContext<E>,
builder_registration_timestamp_override: Option<u64>,
// Used to track unpublished validator registration changes.
validator_registration_cache:
@@ -146,11 +145,11 @@ impl From<ValidatorRegistrationData> for ValidatorRegistrationKey {
}
/// Attempts to produce proposer preparations for all known validators at the beginning of each epoch.
pub struct PreparationService<S, T> {
inner: Arc<Inner<S, T>>,
pub struct PreparationService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<S, T> Clone for PreparationService<S, T> {
impl<T, E: EthSpec> Clone for PreparationService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -158,15 +157,15 @@ impl<S, T> Clone for PreparationService<S, T> {
}
}
impl<S, T> Deref for PreparationService<S, T> {
type Target = Inner<S, T>;
impl<T, E: EthSpec> Deref for PreparationService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> PreparationService<T, E> {
pub fn start_update_service(self, spec: &ChainSpec) -> Result<(), String> {
self.clone().start_validator_registration_service(spec)?;
self.start_proposer_prepare_service(spec)
@@ -177,7 +176,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
let slot_duration = Duration::from_secs(spec.seconds_per_slot);
info!("Proposer preparation service started");
let executor = self.executor.clone();
let executor = self.context.executor.clone();
let spec = spec.clone();
let interval_fut = async move {
@@ -216,13 +215,13 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
let spec = spec.clone();
let slot_duration = Duration::from_secs(spec.seconds_per_slot);
let executor = self.executor.clone();
let executor = self.context.executor.clone();
let validator_registration_fut = async move {
loop {
// Poll the endpoint immediately to ensure fee recipients are received.
if let Err(e) = self.register_validators().await {
error!(error = ?e,"Error during validator registration");
error!(error = ?e, "Error during validator registration");
}
// Wait one slot if the register validator request fails or if we should not publish at the current slot.
@@ -244,9 +243,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
/// This avoids spamming the BN with preparations before the Bellatrix fork epoch, which may
/// cause errors if it doesn't support the preparation API.
fn should_publish_at_current_slot(&self, spec: &ChainSpec) -> bool {
let current_epoch = self.slot_clock.now().map_or(S::E::genesis_epoch(), |slot| {
slot.epoch(S::E::slots_per_epoch())
});
let current_epoch = self
.slot_clock
.now()
.map_or(E::genesis_epoch(), |slot| slot.epoch(E::slots_per_epoch()));
spec.bellatrix_fork_epoch.is_some_and(|fork_epoch| {
current_epoch + PROPOSER_PREPARATION_LOOKAHEAD_EPOCHS >= fork_epoch
})
@@ -367,8 +367,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
// Check if any have changed or it's been `EPOCHS_PER_VALIDATOR_REGISTRATION_SUBMISSION`.
if let Some(slot) = self.slot_clock.now() {
if slot % (S::E::slots_per_epoch() * EPOCHS_PER_VALIDATOR_REGISTRATION_SUBMISSION) == 0
{
if slot % (E::slots_per_epoch() * EPOCHS_PER_VALIDATOR_REGISTRATION_SUBMISSION) == 0 {
self.publish_validator_registration_data(registration_keys)
.await?;
} else if !changed_keys.is_empty() {
@@ -410,7 +409,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
pubkey,
} = key.clone();
let signed_data = match self
match self
.validator_store
.sign_validator_registration_data(ValidatorRegistrationData {
fee_recipient,
@@ -435,13 +434,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
);
continue;
}
};
self.validator_registration_cache
.write()
.insert(key, signed_data.clone());
signed_data
}
};
signed.push(signed_data);
}
@@ -455,10 +448,19 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> PreparationService<S,
})
.await
{
Ok(()) => info!(
count = batch.len(),
"Published validator registrations to the builder network"
),
Ok(()) => {
info!(
count = batch.len(),
"Published validator registrations to the builder network"
);
let mut guard = self.validator_registration_cache.write();
for signed_data in batch {
guard.insert(
ValidatorRegistrationKey::from(signed_data.message.clone()),
signed_data.clone(),
);
}
}
Err(e) => warn!(
error = %e,
"Unable to publish validator registrations to the builder network"

68
validator_client/validator_services/src/sync.rs
View File

@@ -1,13 +1,15 @@
use crate::duties_service::{DutiesService, Error};
use doppelganger_service::DoppelgangerStatus;
use futures::future::join_all;
use logging::crit;
use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
use slot_clock::SlotClock;
use std::collections::{HashMap, HashSet};
use std::marker::PhantomData;
use std::sync::Arc;
use tracing::{debug, info, warn};
use types::{ChainSpec, EthSpec, PublicKeyBytes, Slot, SyncDuty, SyncSelectionProof, SyncSubnetId};
use validator_store::{DoppelgangerStatus, Error as ValidatorStoreError, ValidatorStore};
use validator_store::Error as ValidatorStoreError;
/// Number of epochs in advance to compute selection proofs when not in `distributed` mode.
pub const AGGREGATION_PRE_COMPUTE_EPOCHS: u64 = 2;
@@ -26,11 +28,12 @@ pub const AGGREGATION_PRE_COMPUTE_SLOTS_DISTRIBUTED: u64 = 1;
/// 2. One-at-a-time locking. For the innermost locks on the aggregator duties, all of the functions
/// in this file take care to only lock one validator at a time. We never hold a lock while
/// trying to obtain another one (hence no lock ordering issues).
pub struct SyncDutiesMap {
pub struct SyncDutiesMap<E: EthSpec> {
/// Map from sync committee period to duties for members of that sync committee.
committees: RwLock<HashMap<u64, CommitteeDuties>>,
/// Whether we are in `distributed` mode and using reduced lookahead for aggregate pre-compute.
distributed: bool,
_phantom: PhantomData<E>,
}
/// Duties for a single sync committee period.
@@ -78,11 +81,12 @@ pub struct SlotDuties {
pub aggregators: HashMap<SyncSubnetId, Vec<(u64, PublicKeyBytes, SyncSelectionProof)>>,
}
impl SyncDutiesMap {
impl<E: EthSpec> SyncDutiesMap<E> {
pub fn new(distributed: bool) -> Self {
Self {
committees: RwLock::new(HashMap::new()),
distributed,
_phantom: PhantomData,
}
}
@@ -100,7 +104,7 @@ impl SyncDutiesMap {
}
/// Number of slots in advance to compute selection proofs
fn aggregation_pre_compute_slots<E: EthSpec>(&self) -> u64 {
fn aggregation_pre_compute_slots(&self) -> u64 {
if self.distributed {
AGGREGATION_PRE_COMPUTE_SLOTS_DISTRIBUTED
} else {
@@ -113,7 +117,7 @@ impl SyncDutiesMap {
/// Return the slot up to which proofs should be pre-computed, as well as a vec of
/// `(previous_pre_compute_slot, sync_duty)` pairs for all validators which need to have proofs
/// computed. See `fill_in_aggregation_proofs` for the actual calculation.
fn prepare_for_aggregator_pre_compute<E: EthSpec>(
fn prepare_for_aggregator_pre_compute(
&self,
committee_period: u64,
current_slot: Slot,
@@ -123,7 +127,7 @@ impl SyncDutiesMap {
current_slot,
first_slot_of_period::<E>(committee_period, spec),
);
let pre_compute_lookahead_slots = self.aggregation_pre_compute_slots::<E>();
let pre_compute_lookahead_slots = self.aggregation_pre_compute_slots();
let pre_compute_slot = std::cmp::min(
current_slot + pre_compute_lookahead_slots,
last_slot_of_period::<E>(committee_period, spec),
@@ -183,7 +187,7 @@ impl SyncDutiesMap {
/// Get duties for all validators for the given `wall_clock_slot`.
///
/// This is the entry-point for the sync committee service.
pub fn get_duties_for_slot<E: EthSpec>(
pub fn get_duties_for_slot(
&self,
wall_clock_slot: Slot,
spec: &ChainSpec,
@@ -280,21 +284,21 @@ fn last_slot_of_period<E: EthSpec>(sync_committee_period: u64, spec: &ChainSpec)
first_slot_of_period::<E>(sync_committee_period + 1, spec) - 1
}
pub async fn poll_sync_committee_duties<S: ValidatorStore + 'static, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
) -> Result<(), Error<S::Error>> {
pub async fn poll_sync_committee_duties<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
) -> Result<(), Error> {
let sync_duties = &duties_service.sync_duties;
let spec = &duties_service.spec;
let current_slot = duties_service
.slot_clock
.now()
.ok_or(Error::UnableToReadSlotClock)?;
let current_epoch = current_slot.epoch(S::E::slots_per_epoch());
let current_epoch = current_slot.epoch(E::slots_per_epoch());
// If the Altair fork is yet to be activated, do not attempt to poll for duties.
if spec
.altair_fork_epoch
.map_or(true, |altair_epoch| current_epoch < altair_epoch)
.is_none_or(|altair_epoch| current_epoch < altair_epoch)
{
return Ok(());
}
@@ -313,8 +317,10 @@ pub async fn poll_sync_committee_duties<S: ValidatorStore + 'static, T: SlotCloc
let local_indices = {
let mut local_indices = Vec::with_capacity(local_pubkeys.len());
let vals_ref = duties_service.validator_store.initialized_validators();
let vals = vals_ref.read();
for &pubkey in &local_pubkeys {
if let Some(validator_index) = duties_service.validator_store.validator_index(&pubkey) {
if let Some(validator_index) = vals.get_index(&pubkey) {
local_indices.push(validator_index)
}
}
@@ -336,15 +342,11 @@ pub async fn poll_sync_committee_duties<S: ValidatorStore + 'static, T: SlotCloc
// Pre-compute aggregator selection proofs for the current period.
let (current_pre_compute_slot, new_pre_compute_duties) = sync_duties
.prepare_for_aggregator_pre_compute::<S::E>(
current_sync_committee_period,
current_slot,
spec,
);
.prepare_for_aggregator_pre_compute(current_sync_committee_period, current_slot, spec);
if !new_pre_compute_duties.is_empty() {
let sub_duties_service = duties_service.clone();
duties_service.executor.spawn(
duties_service.context.executor.spawn(
async move {
fill_in_aggregation_proofs(
sub_duties_service,
@@ -377,22 +379,18 @@ pub async fn poll_sync_committee_duties<S: ValidatorStore + 'static, T: SlotCloc
}
// Pre-compute aggregator selection proofs for the next period.
let aggregate_pre_compute_lookahead_slots = sync_duties.aggregation_pre_compute_slots::<S::E>();
let aggregate_pre_compute_lookahead_slots = sync_duties.aggregation_pre_compute_slots();
if (current_slot + aggregate_pre_compute_lookahead_slots)
.epoch(S::E::slots_per_epoch())
.epoch(E::slots_per_epoch())
.sync_committee_period(spec)?
== next_sync_committee_period
{
let (pre_compute_slot, new_pre_compute_duties) = sync_duties
.prepare_for_aggregator_pre_compute::<S::E>(
next_sync_committee_period,
current_slot,
spec,
);
.prepare_for_aggregator_pre_compute(next_sync_committee_period, current_slot, spec);
if !new_pre_compute_duties.is_empty() {
let sub_duties_service = duties_service.clone();
duties_service.executor.spawn(
duties_service.context.executor.spawn(
async move {
fill_in_aggregation_proofs(
sub_duties_service,
@@ -411,11 +409,11 @@ pub async fn poll_sync_committee_duties<S: ValidatorStore + 'static, T: SlotCloc
Ok(())
}
pub async fn poll_sync_committee_duties_for_period<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: &Arc<DutiesService<S, T>>,
pub async fn poll_sync_committee_duties_for_period<T: SlotClock + 'static, E: EthSpec>(
duties_service: &Arc<DutiesService<T, E>>,
local_indices: &[u64],
sync_committee_period: u64,
) -> Result<(), Error<S::Error>> {
) -> Result<(), Error> {
let spec = &duties_service.spec;
// no local validators don't need to poll for sync committee
@@ -473,7 +471,7 @@ pub async fn poll_sync_committee_duties_for_period<S: ValidatorStore, T: SlotClo
.get_mut(&duty.validator_index)
.ok_or(Error::SyncDutiesNotFound(duty.validator_index))?;
let updated = validator_duties.as_ref().map_or(true, |existing_duties| {
let updated = validator_duties.as_ref().is_none_or(|existing_duties| {
let updated_due_to_reorg = existing_duties.duty.validator_sync_committee_indices
!= duty.validator_sync_committee_indices;
if updated_due_to_reorg {
@@ -498,8 +496,8 @@ pub async fn poll_sync_committee_duties_for_period<S: ValidatorStore, T: SlotClo
Ok(())
}
pub async fn fill_in_aggregation_proofs<S: ValidatorStore, T: SlotClock + 'static>(
duties_service: Arc<DutiesService<S, T>>,
pub async fn fill_in_aggregation_proofs<T: SlotClock + 'static, E: EthSpec>(
duties_service: Arc<DutiesService<T, E>>,
pre_compute_duties: &[(Slot, SyncDuty)],
sync_committee_period: u64,
current_slot: Slot,
@@ -521,7 +519,7 @@ pub async fn fill_in_aggregation_proofs<S: ValidatorStore, T: SlotClock + 'stati
continue;
}
let subnet_ids = match duty.subnet_ids::<S::E>() {
let subnet_ids = match duty.subnet_ids::<E>() {
Ok(subnet_ids) => subnet_ids,
Err(e) => {
crit!(
@@ -566,7 +564,7 @@ pub async fn fill_in_aggregation_proofs<S: ValidatorStore, T: SlotClock + 'stati
}
};
match proof.is_aggregator::<S::E>() {
match proof.is_aggregator::<E>() {
Ok(true) => {
debug!(
validator_index = duty.validator_index,

65
validator_client/validator_services/src/sync_committee_service.rs
View File

@@ -1,5 +1,6 @@
use crate::duties_service::DutiesService;
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
use environment::RuntimeContext;
use eth2::types::BlockId;
use futures::future::join_all;
use futures::future::FutureExt;
@@ -9,7 +10,6 @@ use std::collections::HashMap;
use std::ops::Deref;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use task_executor::TaskExecutor;
use tokio::time::{sleep, sleep_until, Duration, Instant};
use tracing::{debug, error, info, trace, warn};
use types::{
@@ -20,11 +20,11 @@ use validator_store::{Error as ValidatorStoreError, ValidatorStore};
pub const SUBSCRIPTION_LOOKAHEAD_EPOCHS: u64 = 4;
pub struct SyncCommitteeService<S: ValidatorStore, T: SlotClock + 'static> {
inner: Arc<Inner<S, T>>,
pub struct SyncCommitteeService<T: SlotClock + 'static, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<S: ValidatorStore, T: SlotClock + 'static> Clone for SyncCommitteeService<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> Clone for SyncCommitteeService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -32,33 +32,33 @@ impl<S: ValidatorStore, T: SlotClock + 'static> Clone for SyncCommitteeService<S
}
}
impl<S: ValidatorStore, T: SlotClock + 'static> Deref for SyncCommitteeService<S, T> {
type Target = Inner<S, T>;
impl<T: SlotClock + 'static, E: EthSpec> Deref for SyncCommitteeService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
pub struct Inner<S: ValidatorStore, T: SlotClock + 'static> {
duties_service: Arc<DutiesService<S, T>>,
validator_store: Arc<S>,
pub struct Inner<T: SlotClock + 'static, E: EthSpec> {
duties_service: Arc<DutiesService<T, E>>,
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T>>,
executor: TaskExecutor,
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
context: RuntimeContext<E>,
/// Boolean to track whether the service has posted subscriptions to the BN at least once.
///
/// This acts as a latch that fires once upon start-up, and then never again.
first_subscription_done: AtomicBool,
}
impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S, T> {
impl<T: SlotClock + 'static, E: EthSpec> SyncCommitteeService<T, E> {
pub fn new(
duties_service: Arc<DutiesService<S, T>>,
validator_store: Arc<S>,
duties_service: Arc<DutiesService<T, E>>,
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T>>,
executor: TaskExecutor,
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
context: RuntimeContext<E>,
) -> Self {
Self {
inner: Arc::new(Inner {
@@ -66,7 +66,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
validator_store,
slot_clock,
beacon_nodes,
executor,
context,
first_subscription_done: AtomicBool::new(false),
}),
}
@@ -80,13 +80,18 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
.spec
.altair_fork_epoch
.and_then(|fork_epoch| {
let current_epoch = self.slot_clock.now()?.epoch(S::E::slots_per_epoch());
let current_epoch = self.slot_clock.now()?.epoch(E::slots_per_epoch());
Some(current_epoch >= fork_epoch)
})
.unwrap_or(false)
}
pub fn start_update_service(self, spec: &ChainSpec) -> Result<(), String> {
if self.duties_service.disable_attesting {
info!("Sync committee service disabled");
return Ok(());
}
let slot_duration = Duration::from_secs(spec.seconds_per_slot);
let duration_to_next_slot = self
.slot_clock
@@ -98,7 +103,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
"Sync committee service started"
);
let executor = self.executor.clone();
let executor = self.context.executor.clone();
let interval_fut = async move {
loop {
@@ -151,7 +156,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
let Some(slot_duties) = self
.duties_service
.sync_duties
.get_duties_for_slot::<S::E>(slot, &self.duties_service.spec)
.get_duties_for_slot(slot, &self.duties_service.spec)
else {
debug!("No duties known for slot {}", slot);
return Ok(());
@@ -197,7 +202,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
// Spawn one task to publish all of the sync committee signatures.
let validator_duties = slot_duties.duties;
let service = self.clone();
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
service
.publish_sync_committee_signatures(slot, block_root, validator_duties)
@@ -209,7 +214,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
let aggregators = slot_duties.aggregators;
let service = self.clone();
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
service
.publish_sync_committee_aggregates(
@@ -311,7 +316,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
) {
for (subnet_id, subnet_aggregators) in aggregators {
let service = self.clone();
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
service
.publish_sync_committee_aggregate_for_subnet(
@@ -349,7 +354,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
};
beacon_node
.get_validator_sync_committee_contribution(&sync_contribution_data)
.get_validator_sync_committee_contribution::<E>(&sync_contribution_data)
.await
})
.await
@@ -435,7 +440,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
fn spawn_subscription_tasks(&self) {
let service = self.clone();
self.inner.executor.spawn(
self.inner.context.executor.spawn(
async move {
service.publish_subscriptions().await.unwrap_or_else(|e| {
error!(
@@ -458,10 +463,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
// At the start of every epoch during the current period, re-post the subscriptions
// to the beacon node. This covers the case where the BN has forgotten the subscriptions
// due to a restart, or where the VC has switched to a fallback BN.
let current_period = sync_period_of_slot::<S::E>(slot, spec)?;
let current_period = sync_period_of_slot::<E>(slot, spec)?;
if !self.first_subscription_done.load(Ordering::Relaxed)
|| slot.as_u64() % S::E::slots_per_epoch() == 0
|| slot.as_u64() % E::slots_per_epoch() == 0
{
duty_slots.push((slot, current_period));
}
@@ -469,9 +474,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
// Near the end of the current period, push subscriptions for the next period to the
// beacon node. We aggressively push every slot in the lead-up, as this is the main way
// that we want to ensure that the BN is subscribed (well in advance).
let lookahead_slot = slot + SUBSCRIPTION_LOOKAHEAD_EPOCHS * S::E::slots_per_epoch();
let lookahead_slot = slot + SUBSCRIPTION_LOOKAHEAD_EPOCHS * E::slots_per_epoch();
let lookahead_period = sync_period_of_slot::<S::E>(lookahead_slot, spec)?;
let lookahead_period = sync_period_of_slot::<E>(lookahead_slot, spec)?;
if lookahead_period > current_period {
duty_slots.push((lookahead_slot, lookahead_period));
@@ -489,7 +494,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
match self
.duties_service
.sync_duties
.get_duties_for_slot::<S::E>(duty_slot, spec)
.get_duties_for_slot(duty_slot, spec)
{
Some(duties) => subscriptions.extend(subscriptions_from_sync_duties(
duties.duties,