gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-31 13:17:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

resolve merge conflict and migrate il service to new pardigmn

Browse Source
This commit is contained in:
Eitan Seri-Levi
2025-05-21 12:43:43 -07:00
parent 7bd50a6fe8 e29b607257
commit e21b578073
358 changed files with 11541 additions and 6759 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
validator_client/src/cli.rs
View File

@@ -67,7 +67,6 @@ pub struct ValidatorClient {
#[clap(
long,
value_name = "SECRETS_DIRECTORY",
conflicts_with = "datadir",
help = "The directory which contains the password to unlock the validator \
voting keypairs. Each password should be contained in a file where the \
name is the 0x-prefixed hex representation of the validators voting public \
@@ -220,6 +219,7 @@ pub struct ValidatorClient {
#[clap(
long,
requires = "http",
value_name = "PORT",
default_value_t = 5062,
help = "Set the listen TCP port for the RESTful HTTP API server.",
@@ -388,7 +388,7 @@ pub struct ValidatorClient {
#[clap(
long,
value_name = "INTEGER",
default_value_t = 30_000_000,
default_value_t = 36_000_000,
requires = "builder_proposals",
help = "The gas limit to be used in all builder proposals for all validators managed \
by this validator client. Note this will not necessarily be used if the gas limit \

2
validator_client/src/config.rs
View File

@@ -10,6 +10,7 @@ use directory::{
use eth2::types::Graffiti;
use graffiti_file::GraffitiFile;
use initialized_validators::Config as InitializedValidatorsConfig;
use lighthouse_validator_store::Config as ValidatorStoreConfig;
use sensitive_url::SensitiveUrl;
use serde::{Deserialize, Serialize};
use std::fs;
@@ -20,7 +21,6 @@ use tracing::{info, warn};
use types::GRAFFITI_BYTES_LEN;
use validator_http_api::{self, PK_FILENAME};
use validator_http_metrics;
use validator_store::Config as ValidatorStoreConfig;
pub const DEFAULT_BEACON_NODE: &str = "http://localhost:5052/";

61
validator_client/src/latency.rs
View File

@@ -1,61 +0,0 @@
use beacon_node_fallback::BeaconNodeFallback;
use environment::RuntimeContext;
use slot_clock::SlotClock;
use std::sync::Arc;
use tokio::time::sleep;
use tracing::debug;
use types::EthSpec;
/// The latency service will run 11/12ths of the way through the slot.
pub const SLOT_DELAY_MULTIPLIER: u32 = 11;
pub const SLOT_DELAY_DENOMINATOR: u32 = 12;
/// Starts a service that periodically checks the latency between the VC and the
/// candidate BNs.
pub fn start_latency_service<T: SlotClock + 'static, E: EthSpec>(
context: RuntimeContext<E>,
slot_clock: T,
beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
) {
let future = async move {
loop {
let sleep_time = slot_clock
.duration_to_next_slot()
.map(|next_slot| {
// This is 11/12ths through the next slot. On mainnet this
// will happen in the 11th second of each slot, one second
// before the next slot.
next_slot + (next_slot / SLOT_DELAY_DENOMINATOR) * SLOT_DELAY_MULTIPLIER
})
// If we can't read the slot clock, just wait one slot. Running
// the measurement at a non-exact time is not a big issue.
.unwrap_or_else(|| slot_clock.slot_duration());
// Sleep until it's time to perform the measurement.
sleep(sleep_time).await;
for (i, measurement) in beacon_nodes.measure_latency().await.iter().enumerate() {
if let Some(latency) = measurement.latency {
debug!(
node = &measurement.beacon_node_id,
latency = latency.as_millis(),
"Measured BN latency"
);
validator_metrics::observe_timer_vec(
&validator_metrics::VC_BEACON_NODE_LATENCY,
&[&measurement.beacon_node_id],
latency,
);
if i == 0 {
validator_metrics::observe_duration(
&validator_metrics::VC_BEACON_NODE_LATENCY_PRIMARY_ENDPOINT,
latency,
);
}
}
}
}
};
context.executor.spawn(future, "latency");
}

131
validator_client/src/lib.rs
View File

@@ -1,7 +1,5 @@
pub mod cli;
pub mod config;
mod latency;
mod notifier;
use crate::cli::ValidatorClient;
pub use config::Config;
@@ -20,14 +18,13 @@ use doppelganger_service::DoppelgangerService;
use environment::RuntimeContext;
use eth2::{reqwest::ClientBuilder, BeaconNodeHttpClient, StatusCode, Timeouts};
use initialized_validators::Error::UnableToOpenVotingKeystore;
use notifier::spawn_notifier;
use lighthouse_validator_store::LighthouseValidatorStore;
use parking_lot::RwLock;
use reqwest::Certificate;
use slot_clock::SlotClock;
use slot_clock::SystemTimeSlotClock;
use std::fs::File;
use std::io::Read;
use std::marker::PhantomData;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
@@ -39,16 +36,18 @@ use tokio::{
use tracing::{debug, error, info, warn};
use types::{EthSpec, Hash256};
use validator_http_api::ApiSecret;
use validator_services::inclusion_list_service::InclusionListServiceBuilder;
use validator_services::notifier_service::spawn_notifier;
use validator_services::{
attestation_service::{AttestationService, AttestationServiceBuilder},
block_service::{BlockService, BlockServiceBuilder},
duties_service::{self, DutiesService},
duties_service::{self, DutiesService, DutiesServiceBuilder},
inclusion_list_service::InclusionListService,
latency_service,
preparation_service::{PreparationService, PreparationServiceBuilder},
sync::SyncDutiesMap,
sync_committee_service::SyncCommitteeService,
};
use validator_store::ValidatorStore;
use validator_store::ValidatorStore as ValidatorStoreTrait;
/// The interval between attempts to contact the beacon node during startup.
const RETRY_DELAY: Duration = Duration::from_secs(2);
@@ -72,24 +71,27 @@ const HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT: u32 = 4;
const HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT: u32 = 4;
const HTTP_GET_VALIDATOR_BLOCK_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_DEFAULT_TIMEOUT_QUOTIENT: u32 = 4;
const DOPPELGANGER_SERVICE_NAME: &str = "doppelganger";
type ValidatorStore<E> = LighthouseValidatorStore<SystemTimeSlotClock, E>;
#[derive(Clone)]
pub struct ProductionValidatorClient<E: EthSpec> {
context: RuntimeContext<E>,
duties_service: Arc<DutiesService<SystemTimeSlotClock, E>>,
block_service: BlockService<SystemTimeSlotClock, E>,
attestation_service: AttestationService<SystemTimeSlotClock, E>,
sync_committee_service: SyncCommitteeService<SystemTimeSlotClock, E>,
inclusion_list_service: InclusionListService<SystemTimeSlotClock, E>,
duties_service: Arc<DutiesService<ValidatorStore<E>, SystemTimeSlotClock>>,
block_service: BlockService<ValidatorStore<E>, SystemTimeSlotClock>,
attestation_service: AttestationService<ValidatorStore<E>, SystemTimeSlotClock>,
sync_committee_service: SyncCommitteeService<ValidatorStore<E>, SystemTimeSlotClock>,
inclusion_list_service: InclusionListService<ValidatorStore<E>, SystemTimeSlotClock>,
doppelganger_service: Option<Arc<DoppelgangerService>>,
preparation_service: PreparationService<SystemTimeSlotClock, E>,
validator_store: Arc<ValidatorStore<SystemTimeSlotClock, E>>,
preparation_service: PreparationService<ValidatorStore<E>, SystemTimeSlotClock>,
validator_store: Arc<ValidatorStore<E>>,
slot_clock: SystemTimeSlotClock,
http_api_listen_addr: Option<SocketAddr>,
config: Config,
beacon_nodes: Arc<BeaconNodeFallback<SystemTimeSlotClock, E>>,
beacon_nodes: Arc<BeaconNodeFallback<SystemTimeSlotClock>>,
genesis_time: u64,
}
@@ -313,6 +315,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
get_debug_beacon_states: slot_duration / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT,
get_deposit_snapshot: slot_duration / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT,
get_validator_block: slot_duration / HTTP_GET_VALIDATOR_BLOCK_TIMEOUT_QUOTIENT,
default: slot_duration / HTTP_DEFAULT_TIMEOUT_QUOTIENT,
}
} else {
Timeouts::set_all(slot_duration.saturating_mul(config.long_timeouts_multiplier))
@@ -374,14 +377,14 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
// Initialize the number of connected, avaliable beacon nodes to 0.
set_gauge(&validator_metrics::AVAILABLE_BEACON_NODES_COUNT, 0);
let mut beacon_nodes: BeaconNodeFallback<_, E> = BeaconNodeFallback::new(
let mut beacon_nodes: BeaconNodeFallback<_> = BeaconNodeFallback::new(
candidates,
config.beacon_node_fallback,
config.broadcast_topics.clone(),
context.eth2_config.spec.clone(),
);
let mut proposer_nodes: BeaconNodeFallback<_, E> = BeaconNodeFallback::new(
let mut proposer_nodes: BeaconNodeFallback<_> = BeaconNodeFallback::new(
proposer_candidates,
config.beacon_node_fallback,
config.broadcast_topics.clone(),
@@ -390,7 +393,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
// Perform some potentially long-running initialization tasks.
let (genesis_time, genesis_validators_root) = tokio::select! {
tuple = init_from_beacon_node(&beacon_nodes, &proposer_nodes) => tuple?,
tuple = init_from_beacon_node::<E>(&beacon_nodes, &proposer_nodes) => tuple?,
() = context.executor.exit() => return Err("Shutting down".to_string())
};
@@ -409,10 +412,10 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
proposer_nodes.set_slot_clock(slot_clock.clone());
let beacon_nodes = Arc::new(beacon_nodes);
start_fallback_updater_service(context.clone(), beacon_nodes.clone())?;
start_fallback_updater_service::<_, E>(context.executor.clone(), beacon_nodes.clone())?;
let proposer_nodes = Arc::new(proposer_nodes);
start_fallback_updater_service(context.clone(), proposer_nodes.clone())?;
start_fallback_updater_service::<_, E>(context.executor.clone(), proposer_nodes.clone())?;
let doppelganger_service = if config.enable_doppelganger_protection {
Some(Arc::new(DoppelgangerService::default()))
@@ -420,7 +423,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
None
};
let validator_store = Arc::new(ValidatorStore::new(
let validator_store = Arc::new(LighthouseValidatorStore::new(
validators,
slashing_protection,
genesis_validators_root,
@@ -446,22 +449,18 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
validator_store.prune_slashing_protection_db(slot.epoch(E::slots_per_epoch()), true);
}
let duties_context = context.service_context("duties".into());
let duties_service = Arc::new(DutiesService {
attesters: <_>::default(),
proposers: <_>::default(),
sync_duties: SyncDutiesMap::new(config.distributed),
inclusion_list_duties: <_>::default(),
slot_clock: slot_clock.clone(),
beacon_nodes: beacon_nodes.clone(),
validator_store: validator_store.clone(),
unknown_validator_next_poll_slots: <_>::default(),
spec: context.eth2_config.spec.clone(),
context: duties_context,
enable_high_validator_count_metrics: config.enable_high_validator_count_metrics,
distributed: config.distributed,
disable_attesting: config.disable_attesting,
});
let duties_service = Arc::new(
DutiesServiceBuilder::new()
.slot_clock(slot_clock.clone())
.beacon_nodes(beacon_nodes.clone())
.validator_store(validator_store.clone())
.spec(context.eth2_config.spec.clone())
.executor(context.executor.clone())
.enable_high_validator_count_metrics(config.enable_high_validator_count_metrics)
.distributed(config.distributed)
.disable_attesting(config.disable_attesting)
.build()?,
);
// Update the metrics server.
if let Some(ctx) = &validator_metrics_ctx {
@@ -473,7 +472,8 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_nodes(beacon_nodes.clone())
.runtime_context(context.service_context("block".into()))
.executor(context.executor.clone())
.chain_spec(context.eth2_config.spec.clone())
.graffiti(config.graffiti)
.graffiti_file(config.graffiti_file.clone());
@@ -489,7 +489,8 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_nodes(beacon_nodes.clone())
.runtime_context(context.service_context("attestation".into()))
.executor(context.executor.clone())
.chain_spec(context.eth2_config.spec.clone())
.disable(config.disable_attesting)
.build()?;
@@ -497,7 +498,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_nodes(beacon_nodes.clone())
.runtime_context(context.service_context("preparation".into()))
.executor(context.executor.clone())
.builder_registration_timestamp_override(config.builder_registration_timestamp_override)
.validator_registration_batch_size(config.validator_registration_batch_size)
.build()?;
@@ -507,16 +508,19 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
validator_store.clone(),
slot_clock.clone(),
beacon_nodes.clone(),
context.service_context("sync_committee".into()),
context.executor.clone(),
);
let inclusion_list_service = InclusionListService::new(
duties_service.clone(),
validator_store.clone(),
slot_clock.clone(),
beacon_nodes.clone(),
context.service_context("inclusion_list".into()),
);
let inclusion_list_service = InclusionListServiceBuilder::new()
.duties_service(duties_service.clone())
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_nodes(beacon_nodes.clone())
.executor(context.executor.clone())
.chain_spec(context.eth2_config.spec.clone())
// TODO(focil) make config driven
.disable(false)
.build()?;
Ok(Self {
context,
@@ -559,12 +563,11 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
config: self.config.http_api.clone(),
sse_logging_components: self.context.sse_logging_components.clone(),
slot_clock: self.slot_clock.clone(),
_phantom: PhantomData,
});
let exit = self.context.executor.exit();
let (listen_addr, server) = validator_http_api::serve(ctx, exit)
let (listen_addr, server) = validator_http_api::serve::<_, E>(ctx, exit)
.map_err(|e| format!("Unable to start HTTP API server: {:?}", e))?;
self.context
@@ -622,11 +625,17 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
info!("Doppelganger protection disabled.")
}
spawn_notifier(self).map_err(|e| format!("Failed to start notifier: {}", e))?;
let context = self.context.service_context("notifier".into());
spawn_notifier(
self.duties_service.clone(),
context.executor,
&self.context.eth2_config.spec,
)
.map_err(|e| format!("Failed to start notifier: {}", e))?;
if self.config.enable_latency_measurement_service {
latency::start_latency_service(
self.context.clone(),
latency_service::start_latency_service(
self.context.executor.clone(),
self.duties_service.slot_clock.clone(),
self.duties_service.beacon_nodes.clone(),
);
@@ -637,12 +646,12 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
}
async fn init_from_beacon_node<E: EthSpec>(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
proposer_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
proposer_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
) -> Result<(u64, Hash256), String> {
loop {
beacon_nodes.update_all_candidates().await;
proposer_nodes.update_all_candidates().await;
beacon_nodes.update_all_candidates::<E>().await;
proposer_nodes.update_all_candidates::<E>().await;
let num_available = beacon_nodes.num_available().await;
let num_total = beacon_nodes.num_total().await;
@@ -719,8 +728,8 @@ async fn init_from_beacon_node<E: EthSpec>(
Ok((genesis.genesis_time, genesis.genesis_validators_root))
}
async fn wait_for_genesis<E: EthSpec>(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
async fn wait_for_genesis(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
genesis_time: u64,
) -> Result<(), String> {
let now = SystemTime::now()
@@ -762,8 +771,8 @@ async fn wait_for_genesis<E: EthSpec>(
/// Request the version from the node, looping back and trying again on failure. Exit once the node
/// has been contacted.
async fn poll_whilst_waiting_for_genesis<E: EthSpec>(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
async fn poll_whilst_waiting_for_genesis(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
genesis_time: Duration,
) -> Result<(), String> {
loop {

150
validator_client/src/notifier.rs
View File

@@ -1,150 +0,0 @@
use crate::{DutiesService, ProductionValidatorClient};
use metrics::set_gauge;
use slot_clock::SlotClock;
use tokio::time::{sleep, Duration};
use tracing::{debug, error, info};
use types::EthSpec;
/// Spawns a notifier service which periodically logs information about the node.
pub fn spawn_notifier<E: EthSpec>(client: &ProductionValidatorClient<E>) -> Result<(), String> {
let context = client.context.service_context("notifier".into());
let executor = context.executor.clone();
let duties_service = client.duties_service.clone();
let slot_duration = Duration::from_secs(context.eth2_config.spec.seconds_per_slot);
let interval_fut = async move {
loop {
if let Some(duration_to_next_slot) = duties_service.slot_clock.duration_to_next_slot() {
sleep(duration_to_next_slot + slot_duration / 2).await;
notify(&duties_service).await;
} else {
error!("Failed to read slot clock");
// If we can't read the slot clock, just wait another slot.
sleep(slot_duration).await;
continue;
}
}
};
executor.spawn(interval_fut, "validator_notifier");
Ok(())
}
/// Performs a single notification routine.
async fn notify<T: SlotClock + 'static, E: EthSpec>(duties_service: &DutiesService<T, E>) {
let (candidate_info, num_available, num_synced) =
duties_service.beacon_nodes.get_notifier_info().await;
let num_total = candidate_info.len();
let num_synced_fallback = num_synced.saturating_sub(1);
set_gauge(
&validator_metrics::AVAILABLE_BEACON_NODES_COUNT,
num_available as i64,
);
set_gauge(
&validator_metrics::SYNCED_BEACON_NODES_COUNT,
num_synced as i64,
);
set_gauge(
&validator_metrics::TOTAL_BEACON_NODES_COUNT,
num_total as i64,
);
if num_synced > 0 {
let primary = candidate_info
.first()
.map(|candidate| candidate.endpoint.as_str())
.unwrap_or("None");
info!(
primary,
total = num_total,
available = num_available,
synced = num_synced,
"Connected to beacon node(s)"
)
} else {
error!(
total = num_total,
available = num_available,
synced = num_synced,
"No synced beacon nodes"
)
}
if num_synced_fallback > 0 {
set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 1);
} else {
set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 0);
}
for info in candidate_info {
if let Ok(health) = info.health {
debug!(
status = "Connected",
index = info.index,
endpoint = info.endpoint,
head_slot = %health.head,
is_optimistic = ?health.optimistic_status,
execution_engine_status = ?health.execution_status,
health_tier = %health.health_tier,
"Beacon node info"
);
} else {
debug!(
status = "Disconnected",
index = info.index,
endpoint = info.endpoint,
"Beacon node info"
);
}
}
if let Some(slot) = duties_service.slot_clock.now() {
let epoch = slot.epoch(E::slots_per_epoch());
let total_validators = duties_service.total_validator_count();
let proposing_validators = duties_service.proposer_count(epoch);
let attesting_validators = duties_service.attester_count(epoch);
let doppelganger_detecting_validators = duties_service.doppelganger_detecting_count();
if doppelganger_detecting_validators > 0 {
info!(
doppelganger_detecting_validators,
"Listening for doppelgangers"
)
}
if total_validators == 0 {
info!(
msg = "see `lighthouse vm create --help` or the HTTP API documentation",
"No validators present"
)
} else if total_validators == attesting_validators {
info!(
current_epoch_proposers = proposing_validators,
active_validators = attesting_validators,
total_validators = total_validators,
%epoch,
%slot,
"All validators active"
);
} else if attesting_validators > 0 {
info!(
current_epoch_proposers = proposing_validators,
active_validators = attesting_validators,
total_validators = total_validators,
%epoch,
%slot,
"Some validators active"
);
} else {
info!(
validators = total_validators,
%epoch,
%slot,
"Awaiting activation"
);
}
} else {
error!("Unable to read slot clock");
}
}