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

Merge branch 'dvt' into into-anchor

Browse Source
This commit is contained in:
Daniel Knopik
2025-07-15 15:52:26 +02:00
parent 9b84dac6d6 b5e523c806
commit 701c199ee1
285 changed files with 9820 additions and 11712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
validator_client/beacon_node_fallback/Cargo.toml
View File

@@ -13,6 +13,7 @@ clap = { workspace = true }
eth2 = { workspace = true }
futures = { workspace = true }
itertools = { workspace = true }
sensitive_url = { workspace = true }
serde = { workspace = true }
slot_clock = { workspace = true }
strum = { workspace = true }

36
validator_client/beacon_node_fallback/src/lib.rs
View File

@@ -8,8 +8,9 @@ use beacon_node_health::{
IsOptimistic, SyncDistanceTier,
};
use clap::ValueEnum;
use eth2::BeaconNodeHttpClient;
use eth2::{BeaconNodeHttpClient, Timeouts};
use futures::future;
use sensitive_url::SensitiveUrl;
use serde::{ser::SerializeStruct, Deserialize, Serialize, Serializer};
use slot_clock::SlotClock;
use std::cmp::Ordering;
@@ -455,6 +456,39 @@ impl<T: SlotClock> BeaconNodeFallback<T> {
(candidate_info, num_available, num_synced)
}
/// Update the list of candidates with a new list.
/// Returns `Ok(new_list)` if the update was successful.
/// Returns `Err(some_err)` if the list is empty.
pub async fn update_candidates_list(
&self,
new_list: Vec<SensitiveUrl>,
use_long_timeouts: bool,
) -> Result<Vec<SensitiveUrl>, String> {
if new_list.is_empty() {
return Err("list cannot be empty".to_string());
}
let timeouts: Timeouts = if new_list.len() == 1 || use_long_timeouts {
Timeouts::set_all(Duration::from_secs(self.spec.seconds_per_slot))
} else {
Timeouts::use_optimized_timeouts(Duration::from_secs(self.spec.seconds_per_slot))
};
let new_candidates: Vec<CandidateBeaconNode> = new_list
.clone()
.into_iter()
.enumerate()
.map(|(index, url)| {
CandidateBeaconNode::new(BeaconNodeHttpClient::new(url, timeouts.clone()), index)
})
.collect();
let mut candidates = self.candidates.write().await;
*candidates = new_candidates;
Ok(new_list)
}
/// Loop through ALL candidates in `self.candidates` and update their sync status.
///
/// It is possible for a node to return an unsynced status while continuing to serve

64
validator_client/http_api/src/lib.rs
View File

@@ -22,6 +22,7 @@ use account_utils::{
};
pub use api_secret::ApiSecret;
use beacon_node_fallback::CandidateInfo;
use core::convert::Infallible;
use create_validator::{
create_validators_mnemonic, create_validators_web3signer, get_voting_password_storage,
};
@@ -30,7 +31,7 @@ use eth2::lighthouse_vc::{
std_types::{AuthResponse, GetFeeRecipientResponse, GetGasLimitResponse},
types::{
self as api_types, GenericResponse, GetGraffitiResponse, Graffiti, PublicKey,
PublicKeyBytes, SetGraffitiRequest,
PublicKeyBytes, SetGraffitiRequest, UpdateCandidatesRequest, UpdateCandidatesResponse,
},
};
use health_metrics::observe::Observe;
@@ -38,6 +39,7 @@ use lighthouse_version::version_with_platform;
use logging::crit;
use logging::SSELoggingComponents;
use parking_lot::RwLock;
use sensitive_url::SensitiveUrl;
use serde::{Deserialize, Serialize};
use slot_clock::SlotClock;
use std::collections::HashMap;
@@ -53,7 +55,8 @@ use tracing::{info, warn};
use types::{ChainSpec, ConfigAndPreset, EthSpec};
use validator_dir::Builder as ValidatorDirBuilder;
use validator_services::block_service::BlockService;
use warp::{sse::Event, Filter};
use warp::{reply::Response, sse::Event, Filter};
use warp_utils::reject::convert_rejection;
use warp_utils::task::blocking_json_task;
#[derive(Debug)]
@@ -102,6 +105,7 @@ pub struct Config {
pub allow_keystore_export: bool,
pub store_passwords_in_secrets_dir: bool,
pub http_token_path: PathBuf,
pub bn_long_timeouts: bool,
}
impl Default for Config {
@@ -121,6 +125,7 @@ impl Default for Config {
allow_keystore_export: false,
store_passwords_in_secrets_dir: false,
http_token_path,
bn_long_timeouts: false,
}
}
}
@@ -147,6 +152,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
let config = &ctx.config;
let allow_keystore_export = config.allow_keystore_export;
let store_passwords_in_secrets_dir = config.store_passwords_in_secrets_dir;
let use_long_timeouts = config.bn_long_timeouts;
// Configure CORS.
let cors_builder = {
@@ -839,6 +845,59 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
})
});
// POST /lighthouse/beacon/update
let post_lighthouse_beacon_update = warp::path("lighthouse")
.and(warp::path("beacon"))
.and(warp::path("update"))
.and(warp::path::end())
.and(warp::body::json())
.and(block_service_filter.clone())
.then(
move |request: UpdateCandidatesRequest,
block_service: BlockService<LighthouseValidatorStore<T, E>, T>| async move {
async fn parse_urls(urls: &[String]) -> Result<Vec<SensitiveUrl>, Response> {
match urls
.iter()
.map(|url| SensitiveUrl::parse(url).map_err(|e| e.to_string()))
.collect()
{
Ok(sensitive_urls) => Ok(sensitive_urls),
Err(_) => Err(convert_rejection::<Infallible>(Err(
warp_utils::reject::custom_bad_request(
"one or more urls could not be parsed".to_string(),
),
))
.await),
}
}
let beacons: Vec<SensitiveUrl> = match parse_urls(&request.beacon_nodes).await {
Ok(new_beacons) => {
match block_service
.beacon_nodes
.update_candidates_list(new_beacons, use_long_timeouts)
.await
{
Ok(beacons) => beacons,
Err(e) => {
return convert_rejection::<Infallible>(Err(
warp_utils::reject::custom_bad_request(e.to_string()),
))
.await
}
}
}
Err(e) => return e,
};
let response: UpdateCandidatesResponse = UpdateCandidatesResponse {
new_beacon_nodes_list: beacons.iter().map(|surl| surl.to_string()).collect(),
};
blocking_json_task(move || Ok(api_types::GenericResponse::from(response))).await
},
);
// Standard key-manager endpoints.
let eth_v1 = warp::path("eth").and(warp::path("v1"));
let std_keystores = eth_v1.and(warp::path("keystores")).and(warp::path::end());
@@ -1316,6 +1375,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
.or(post_std_keystores)
.or(post_std_remotekeys)
.or(post_graffiti)
.or(post_lighthouse_beacon_update)
.recover(warp_utils::reject::handle_rejection),
))
.or(warp::patch()

1
validator_client/http_api/src/test_utils.rs
View File

@@ -173,6 +173,7 @@ impl ApiTester {
allow_keystore_export: true,
store_passwords_in_secrets_dir: false,
http_token_path: tempdir().unwrap().path().join(PK_FILENAME),
bn_long_timeouts: false,
}
}

1
validator_client/http_api/src/tests.rs
View File

@@ -126,6 +126,7 @@ impl ApiTester {
allow_keystore_export: true,
store_passwords_in_secrets_dir: false,
http_token_path: token_path,
bn_long_timeouts: false,
},
sse_logging_components: None,
slot_clock: slot_clock.clone(),

2
validator_client/initialized_validators/src/lib.rs
View File

@@ -159,7 +159,7 @@ pub struct InitializedValidator {
impl InitializedValidator {
/// Return a reference to this validator's lockfile if it has one.
pub fn keystore_lockfile(&self) -> Option<MappedMutexGuard<Lockfile>> {
pub fn keystore_lockfile(&self) -> Option<MappedMutexGuard<'_, Lockfile>> {
match self.signing_method.as_ref() {
SigningMethod::LocalKeystore {
ref voting_keystore_lockfile,

99
validator_client/src/lib.rs
View File

@@ -54,24 +54,6 @@ const RETRY_DELAY: Duration = Duration::from_secs(2);
/// The time between polls when waiting for genesis.
const WAITING_FOR_GENESIS_POLL_TIME: Duration = Duration::from_secs(12);
/// Specific timeout constants for HTTP requests involved in different validator duties.
/// This can help ensure that proper endpoint fallback occurs.
const HTTP_ATTESTATION_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_ATTESTATION_SUBSCRIPTIONS_TIMEOUT_QUOTIENT: u32 = 24;
const HTTP_ATTESTATION_AGGREGATOR_TIMEOUT_QUOTIENT: u32 = 24; // For DVT involving middleware only
const HTTP_LIVENESS_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_PROPOSAL_TIMEOUT_QUOTIENT: u32 = 2;
const HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_SYNC_COMMITTEE_CONTRIBUTION_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_SYNC_AGGREGATOR_TIMEOUT_QUOTIENT: u32 = 24; // For DVT involving middleware only
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";
/// Compute attestation selection proofs this many slots before they are required.
@@ -105,7 +87,6 @@ pub struct ProductionValidatorClient<E: EthSpec> {
slot_clock: SystemTimeSlotClock,
http_api_listen_addr: Option<SocketAddr>,
config: Config,
beacon_nodes: Arc<BeaconNodeFallback<SystemTimeSlotClock>>,
genesis_time: u64,
}
@@ -310,27 +291,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
// Use quicker timeouts if a fallback beacon node exists.
let timeouts = if i < last_beacon_node_index && !config.use_long_timeouts {
info!("Fallback endpoints are available, using optimized timeouts.");
Timeouts {
attestation: slot_duration / HTTP_ATTESTATION_TIMEOUT_QUOTIENT,
attester_duties: slot_duration / HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT,
attestation_subscriptions: slot_duration
/ HTTP_ATTESTATION_SUBSCRIPTIONS_TIMEOUT_QUOTIENT,
attestation_aggregators: slot_duration
/ HTTP_ATTESTATION_AGGREGATOR_TIMEOUT_QUOTIENT,
liveness: slot_duration / HTTP_LIVENESS_TIMEOUT_QUOTIENT,
proposal: slot_duration / HTTP_PROPOSAL_TIMEOUT_QUOTIENT,
proposer_duties: slot_duration / HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT,
sync_committee_contribution: slot_duration
/ HTTP_SYNC_COMMITTEE_CONTRIBUTION_TIMEOUT_QUOTIENT,
sync_duties: slot_duration / HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT,
sync_aggregators: slot_duration / HTTP_SYNC_AGGREGATOR_TIMEOUT_QUOTIENT,
get_beacon_blocks_ssz: slot_duration
/ HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT,
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,
}
Timeouts::use_optimized_timeouts(slot_duration)
} else {
Timeouts::set_all(slot_duration.saturating_mul(config.long_timeouts_multiplier))
};
@@ -574,7 +535,6 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
slot_clock,
http_api_listen_addr: None,
genesis_time,
beacon_nodes,
})
}
@@ -620,7 +580,7 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
};
// Wait until genesis has occurred.
wait_for_genesis(&self.beacon_nodes, self.genesis_time).await?;
wait_for_genesis(self.genesis_time).await?;
duties_service::start_update_service(self.duties_service.clone(), block_service_tx);
@@ -761,10 +721,7 @@ async fn init_from_beacon_node<E: EthSpec>(
Ok((genesis.genesis_time, genesis.genesis_validators_root))
}
async fn wait_for_genesis(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
genesis_time: u64,
) -> Result<(), String> {
async fn wait_for_genesis(genesis_time: u64) -> Result<(), String> {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|e| format!("Unable to read system time: {:?}", e))?;
@@ -784,7 +741,7 @@ async fn wait_for_genesis(
// Start polling the node for pre-genesis information, cancelling the polling as soon as the
// timer runs out.
tokio::select! {
result = poll_whilst_waiting_for_genesis(beacon_nodes, genesis_time) => result?,
result = poll_whilst_waiting_for_genesis(genesis_time) => result?,
() = sleep(genesis_time - now) => ()
};
@@ -804,46 +761,20 @@ async fn wait_for_genesis(
/// 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(
beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock>,
genesis_time: Duration,
) -> Result<(), String> {
async fn poll_whilst_waiting_for_genesis(genesis_time: Duration) -> Result<(), String> {
loop {
match beacon_nodes
.first_success(|beacon_node| async move { beacon_node.get_lighthouse_staking().await })
.await
{
Ok(is_staking) => {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|e| format!("Unable to read system time: {:?}", e))?;
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|e| format!("Unable to read system time: {:?}", e))?;
if !is_staking {
error!(
msg = "this will caused missed duties",
info = "see the --staking CLI flag on the beacon node",
"Staking is disabled for beacon node"
);
}
if now < genesis_time {
info!(
bn_staking_enabled = is_staking,
seconds_to_wait = (genesis_time - now).as_secs(),
"Waiting for genesis"
);
} else {
break Ok(());
}
}
Err(e) => {
error!(
error = %e,
"Error polling beacon node"
);
}
if now < genesis_time {
info!(
seconds_to_wait = (genesis_time - now).as_secs(),
"Waiting for genesis"
);
} else {
break Ok(());
}
sleep(WAITING_FOR_GENESIS_POLL_TIME).await;
}
}

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

@@ -1,6 +1,5 @@
use crate::duties_service::{DutiesService, DutyAndProof};
use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
use either::Either;
use futures::future::join_all;
use logging::crit;
use slot_clock::SlotClock;
@@ -461,40 +460,32 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
&validator_metrics::ATTESTATION_SERVICE_TIMES,
&[validator_metrics::ATTESTATIONS_HTTP_POST],
);
if fork_name.electra_enabled() {
let single_attestations = attestations
.iter()
.zip(validator_indices)
.filter_map(|(a, i)| {
match a.to_single_attestation_with_attester_index(*i) {
Ok(a) => Some(a),
Err(e) => {
// This shouldn't happen unless BN and VC are out of sync with
// respect to the Electra fork.
error!(
error = ?e,
committee_index = attestation_data.index,
slot = slot.as_u64(),
"type" = "unaggregated",
"Unable to convert to SingleAttestation"
);
None
}
}
})
.collect::<Vec<_>>();
beacon_node
.post_beacon_pool_attestations_v2::<S::E>(
Either::Right(single_attestations),
fork_name,
)
.await
} else {
beacon_node
.post_beacon_pool_attestations_v1(attestations)
.await
}
let single_attestations = attestations
.iter()
.zip(validator_indices)
.filter_map(|(a, i)| {
match a.to_single_attestation_with_attester_index(*i) {
Ok(a) => Some(a),
Err(e) => {
// This shouldn't happen unless BN and VC are out of sync with
// respect to the Electra fork.
error!(
error = ?e,
committee_index = attestation_data.index,
slot = slot.as_u64(),
"type" = "unaggregated",
"Unable to convert to SingleAttestation"
);
None
}
}
})
.collect::<Vec<_>>();
beacon_node
.post_beacon_pool_attestations_v2::<S::E>(single_attestations, fork_name)
.await
})
.await
{

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

@@ -595,8 +595,12 @@ pub async fn fill_in_aggregation_proofs<S: ValidatorStore, T: SlotClock + 'stati
current_slot: Slot,
pre_compute_slot: Slot,
) {
// Start at the next slot, as aggregation proofs for the duty at the current slot are no longer
// required since we do the actual aggregation in the slot before the duty slot.
let start_slot = current_slot.as_u64() + 1;
// Generate selection proofs for each validator at each slot, one slot at a time.
for slot in ((current_slot.as_u64() + 1)..=(pre_compute_slot.as_u64() + 1)).map(Slot::new) {
for slot in (start_slot..=pre_compute_slot.as_u64()).map(Slot::new) {
// For distributed mode
if duties_service
.sync_duties