diff --git a/common/eth2/src/lib.rs b/common/eth2/src/lib.rs index a02b948a909..e823ffd2e8b 100644 --- a/common/eth2/src/lib.rs +++ b/common/eth2/src/lib.rs @@ -82,6 +82,7 @@ const HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT: u32 = 4; const HTTP_SYNC_AGGREGATOR_TIMEOUT_QUOTIENT: u32 = 24; // For DVT involving middleware only // TODO(EIP-7732): Determine what this quotient should be const HTTP_PTC_DUTIES_TIMEOUT_QUOTIENT: u32 = 4; +const HTTP_INCLUSION_LIST_DUTIES_TIMEOUT_QUOTIENT: u32 = 4; 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; @@ -104,6 +105,7 @@ pub struct Timeouts { pub sync_duties: Duration, pub sync_aggregators: Duration, pub ptc_duties: Duration, + pub inclusion_list_duties: Duration, pub get_beacon_blocks_ssz: Duration, pub get_debug_beacon_states: Duration, pub get_deposit_snapshot: Duration, @@ -126,6 +128,7 @@ impl Timeouts { sync_duties: timeout, sync_aggregators: timeout, ptc_duties: timeout, + inclusion_list_duties: timeout, get_beacon_blocks_ssz: timeout, get_debug_beacon_states: timeout, get_deposit_snapshot: timeout, @@ -150,6 +153,7 @@ impl Timeouts { sync_duties: base_timeout / HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT, sync_aggregators: base_timeout / HTTP_SYNC_AGGREGATOR_TIMEOUT_QUOTIENT, ptc_duties: base_timeout / HTTP_PTC_DUTIES_TIMEOUT_QUOTIENT, + inclusion_list_duties: base_timeout / HTTP_INCLUSION_LIST_DUTIES_TIMEOUT_QUOTIENT, get_beacon_blocks_ssz: base_timeout / HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT, get_debug_beacon_states: base_timeout / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT, get_deposit_snapshot: base_timeout / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT, @@ -4133,6 +4137,29 @@ impl BeaconNodeHttpClient { ) .await } + + /// `POST validator/duties/inclusion_list/{epoch}` + pub async fn post_validator_duties_inclusion_list( + &self, + epoch: Epoch, + indices: &[u64], + ) -> Result>, Error> { + let mut path = self.eth_path(V1)?; + + path.path_segments_mut() + .map_err(|()| Error::InvalidUrl(self.server.clone()))? + .push("validator") + .push("duties") + .push("inclusion_list") + .push(&epoch.to_string()); + + self.post_with_timeout_and_response( + path, + &ValidatorIndexDataRef(indices), + self.timeouts.inclusion_list_duties, + ) + .await + } } #[cfg(test)] diff --git a/common/eth2/src/types.rs b/common/eth2/src/types.rs index 08dc2e00c5f..eedcfc8ff30 100644 --- a/common/eth2/src/types.rs +++ b/common/eth2/src/types.rs @@ -863,6 +863,14 @@ pub struct PtcDuty { pub slot: Slot, } +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct InclusionListDuty { + pub pubkey: PublicKeyBytes, + #[serde(with = "serde_utils::quoted_u64")] + pub validator_index: u64, + pub slot: Slot, +} + #[derive(Clone, Deserialize)] pub struct ValidatorBlocksQuery { pub randao_reveal: SignatureBytes, diff --git a/validator_client/http_metrics/src/lib.rs b/validator_client/http_metrics/src/lib.rs index a4802158682..594242dbbf8 100644 --- a/validator_client/http_metrics/src/lib.rs +++ b/validator_client/http_metrics/src/lib.rs @@ -205,6 +205,16 @@ pub fn gather_prometheus_metrics( &[NEXT_EPOCH], duties_service.ptc_count(next_epoch) as i64, ); + set_int_gauge( + &IL_COUNT, + &[CURRENT_EPOCH], + duties_service.il_duties_count(current_epoch) as i64, + ); + set_int_gauge( + &IL_COUNT, + &[NEXT_EPOCH], + duties_service.il_duties_count(next_epoch) as i64, + ); } } diff --git a/validator_client/validator_metrics/src/lib.rs b/validator_client/validator_metrics/src/lib.rs index 46a86381f91..812140d018a 100644 --- a/validator_client/validator_metrics/src/lib.rs +++ b/validator_client/validator_metrics/src/lib.rs @@ -28,6 +28,11 @@ pub const UPDATE_PTC_FETCH: &str = "update_ptc_fetch"; pub const UPDATE_PTC_STORE: &str = "update_ptc_store"; pub const ATTESTER_DUTIES_HTTP_POST: &str = "attester_duties_http_post"; pub const PTC_DUTIES_HTTP_POST: &str = "ptc_duties_http_post"; +pub const UPDATE_IL_DUTIES_CURRENT_EPOCH: &str = "update_il_duties_current_epoch"; +pub const UPDATE_IL_DUTIES_NEXT_EPOCH: &str = "update_il_duties_next_epoch"; +pub const UPDATE_IL_DUTIES_FETCH: &str = "update_il_duties_fetch"; +pub const UPDATE_IL_DUTIES_STORE: &str = "update_il_duties_store"; +pub const IL_DUTIES_HTTP_POST: &str = "il_duties_http_post"; pub const PROPOSER_DUTIES_HTTP_GET: &str = "proposer_duties_http_get"; pub const VALIDATOR_DUTIES_SYNC_HTTP_POST: &str = "validator_duties_sync_http_post"; pub const VALIDATOR_ID_HTTP_GET: &str = "validator_id_http_get"; @@ -174,6 +179,13 @@ pub static PTC_COUNT: LazyLock> = LazyLock::new(|| { &["task"], ) }); +pub static IL_COUNT: LazyLock> = LazyLock::new(|| { + try_create_int_gauge_vec( + "vc_beacon_il_count", + "Number of IL (Inclusion List Committee) validators on this host", + &["task"], + ) +}); pub static PROPOSAL_CHANGED: LazyLock> = LazyLock::new(|| { try_create_int_counter( "vc_beacon_block_proposal_changed", diff --git a/validator_client/validator_services/src/duties_service.rs b/validator_client/validator_services/src/duties_service.rs index 5fe413a2166..64add327bff 100644 --- a/validator_client/validator_services/src/duties_service.rs +++ b/validator_client/validator_services/src/duties_service.rs @@ -13,7 +13,7 @@ use beacon_node_fallback::{ApiTopic, BeaconNodeFallback}; use bls::PublicKeyBytes; use eth2::types::{ AttesterData, BeaconCommitteeSelection, BeaconCommitteeSubscription, DutiesResponse, - ProposerData, PtcDuty, StateId, ValidatorId, + InclusionListDuty, ProposerData, PtcDuty, StateId, ValidatorId, }; use futures::{ StreamExt, @@ -47,6 +47,7 @@ const VALIDATOR_METRICS_MIN_COUNT: usize = 64; /// reduces the amount of data that needs to be transferred. const INITIAL_DUTIES_QUERY_SIZE: usize = 1; const INITIAL_PTC_DUTIES_QUERY_SIZE: usize = 1; +const INITIAL_IL_DUTIES_QUERY_SIZE: usize = 1; /// Offsets from the attestation duty slot at which a subscription should be sent. const ATTESTATION_SUBSCRIPTION_OFFSETS: [u64; 8] = [3, 4, 5, 6, 7, 8, 16, 32]; @@ -85,6 +86,7 @@ pub enum Error { UnableToReadSlotClock, FailedToDownloadAttesters(#[allow(dead_code)] String), FailedToDownloadPtc(#[allow(dead_code)] String), + FailedToDownloadILs(#[allow(dead_code)] String), FailedToProduceSelectionProof(#[allow(dead_code)] ValidatorStoreError), InvalidModulo(#[allow(dead_code)] ArithError), Arith(#[allow(dead_code)] ArithError), @@ -286,6 +288,7 @@ type DependentRoot = Hash256; type AttesterMap = HashMap>; type ProposerMap = HashMap)>; type PtcMap = HashMap)>; +type IlMap = HashMap)>; pub struct DutiesServiceBuilder { /// Provides the canonical list of locally-managed validators. @@ -395,6 +398,7 @@ impl DutiesServiceBuilder { proposers: Default::default(), sync_duties: SyncDutiesMap::new(self.sync_selection_proof_config), ptc_duties: Default::default(), + il_duties: Default::default(), validator_store: self .validator_store .ok_or("Cannot build DutiesService without validator_store")?, @@ -428,6 +432,8 @@ pub struct DutiesService { pub sync_duties: SyncDutiesMap, /// Maps an epoch to PTC duties for locally-managed validators. pub ptc_duties: RwLock, + /// Maps an epoch to the IL committee duties for locally-managed validators. + pub il_duties: RwLock, /// Provides the canonical list of locally-managed validators. pub validator_store: Arc, /// Maps unknown validator pubkeys to the next slot time when a poll should be conducted again. @@ -500,6 +506,15 @@ impl DutiesService { .unwrap_or(0) } + /// Returns the total number of validators that have IL duties in the given epoch. + pub fn il_duties_count(&self, epoch: Epoch) -> usize { + self.il_duties + .read() + .get(&epoch) + .map(|(_, duties)| duties.len()) + .unwrap_or(0) + } + /// Returns the total number of validators that are in a doppelganger detection period. pub fn doppelganger_detecting_count(&self) -> usize { self.validator_store @@ -581,6 +596,32 @@ impl DutiesService { }) .unwrap_or_default() } + + /// Get IL committee duties for a specific slot. + /// + /// Returns the epoch's dependent root alongside the duties for local validators who have + /// IL committee assignments at the given slot, or `None` if the duties for the slot's epoch + /// have not been downloaded yet. + pub fn get_il_duties_for_slot( + &self, + slot: Slot, + ) -> Option<(DependentRoot, Vec)> { + let epoch = slot.epoch(S::E::slots_per_epoch()); + + self.il_duties + .read() + .get(&epoch) + .map(|(dependent_root, il_duties)| { + ( + *dependent_root, + il_duties + .iter() + .filter(|il_duty| il_duty.slot == slot) + .cloned() + .collect(), + ) + }) + } } /// Start the service that periodically polls the beacon node for validator duties. This will start @@ -764,6 +805,54 @@ pub fn start_update_service "duties_service_ptc", ); } + + // Spawn the task which keeps track of the inclusion list committee duties. + // Only track IL committee duties if the heze fork is scheduled + if core_duties_service.spec.is_heze_scheduled() { + let duties_service = core_duties_service.clone(); + core_duties_service.executor.spawn( + async move { + loop { + let Some(current_slot) = duties_service.slot_clock.now() else { + // Sleep for one slot if we are unable to read from the system clock + sleep(duties_service.slot_clock.slot_duration()).await; + continue; + }; + + let current_epoch = current_slot.epoch(S::E::slots_per_epoch()); + let Some(heze_fork_epoch) = duties_service.spec.heze_fork_epoch else { + // Heze fork epoch not configured + break; + }; + + if current_epoch + 1 < heze_fork_epoch { + // Wait until the next slot and check again if the Heze fork epoch is close + if let Some(duration) = duties_service.slot_clock.duration_to_next_slot() { + sleep(duration).await; + } else { + sleep(duties_service.slot_clock.slot_duration()).await; + } + continue; + } + + if let Err(e) = poll_beacon_il_committee_duties(&duties_service).await { + error!( + error = ?e, + "Failed to poll il committee duties" + ) + } + + if let Some(duration) = duties_service.slot_clock.duration_to_next_slot() { + sleep(duration).await; + } else { + // Sleep for one slot if we are unable to read from the system clock + sleep(duties_service.slot_clock.slot_duration()).await; + } + } + }, + "duties_service_il_committee", + ) + } } /// Iterate through all the voting pubkeys in the `ValidatorStore` and attempt to learn any unknown @@ -1404,6 +1493,26 @@ async fn post_validator_duties_ptc( .map_err(|e| Error::FailedToDownloadPtc(e.to_string())) } +async fn post_validator_il_duties( + duties_service: &Arc>, + epoch: Epoch, + validator_indices: &[u64], +) -> Result>, Error> { + duties_service + .beacon_nodes + .first_success(|beacon_node| async move { + let _timer = validator_metrics::start_timer_vec( + &validator_metrics::DUTIES_SERVICE_TIMES, + &[validator_metrics::IL_DUTIES_HTTP_POST], + ); + beacon_node + .post_validator_duties_inclusion_list(epoch, validator_indices) + .await + }) + .await + .map_err(|e| Error::FailedToDownloadILs(e.to_string())) +} + /// Compute the attestation selection proofs for the `duties` and add them to the `attesters` map. /// /// Duties are computed in batches each slot. If a re-org is detected then the process will @@ -1984,6 +2093,189 @@ async fn poll_beacon_ptc_attesters_for_epoch< Ok(()) } +async fn poll_beacon_il_committee_duties( + duties_service: &Arc>, +) -> Result<(), Error> { + let current_epoch_timer = validator_metrics::start_timer_vec( + &validator_metrics::DUTIES_SERVICE_TIMES, + &[validator_metrics::UPDATE_IL_DUTIES_CURRENT_EPOCH], + ); + + let current_slot = duties_service + .slot_clock + .now() + .ok_or(Error::UnableToReadSlotClock)?; + let current_epoch = current_slot.epoch(S::E::slots_per_epoch()); + + // Collect *all* pubkeys, even those undergoing doppelganger protection. + let local_pubkeys: HashSet = duties_service + .validator_store + .voting_pubkeys(DoppelgangerStatus::ignored); + let local_indices = local_pubkeys + .iter() + .filter_map(|pubkey| duties_service.validator_store.validator_index(pubkey)) + .collect::>(); + + // Poll for current epoch + if let Err(e) = poll_beacon_il_duties_for_epoch( + duties_service, + current_epoch, + &local_indices, + &local_pubkeys, + ) + .await + { + error!( + %current_epoch, + request_epoch = %current_epoch, + err = ?e, + "Failed to download IL committee duties" + ); + } + drop(current_epoch_timer); + + let next_epoch_timer = validator_metrics::start_timer_vec( + &validator_metrics::DUTIES_SERVICE_TIMES, + &[validator_metrics::UPDATE_IL_DUTIES_NEXT_EPOCH], + ); + + // Poll for next epoch + let next_epoch = current_epoch + 1; + if let Err(e) = + poll_beacon_il_duties_for_epoch(duties_service, next_epoch, &local_indices, &local_pubkeys) + .await + { + error!( + %current_epoch, + request_epoch = %next_epoch, + err = ?e, + "Failed to download IL committee duties" + ); + } + drop(next_epoch_timer); + + // Prune old IL committee duties + duties_service + .il_duties + .write() + .retain(|&epoch, _| epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch); + + Ok(()) +} + +/// For the given `local_indices` and `local_pubkeys`, download the IL committee duties +/// for the given `epoch` and store them in `duties_service.il_duties` using bandwidth optimization. +async fn poll_beacon_il_duties_for_epoch( + duties_service: &Arc>, + epoch: Epoch, + local_indices: &[u64], + local_pubkeys: &HashSet, +) -> Result<(), Error> { + if local_indices.is_empty() { + debug!( + %epoch, + "No validators, not downloading IL duties" + ); + return Ok(()); + } + + let fetch_timer = validator_metrics::start_timer_vec( + &validator_metrics::DUTIES_SERVICE_TIMES, + &[validator_metrics::UPDATE_IL_DUTIES_FETCH], + ); + + // TODO(heze) Same limitation as PTC duties: + // only `dependent_root` changes are detected, so validators added mid-epoch won't get IL duties + // until the next epoch boundary. + let initial_indices_to_request = + &local_indices[0..min(INITIAL_IL_DUTIES_QUERY_SIZE, local_indices.len())]; + + let response = + post_validator_il_duties(duties_service, epoch, initial_indices_to_request).await?; + let dependent_root = response.dependent_root; + + // Check if we need to update duties for this epoch and collect validators to update. + // We update if we have no epoch data OR if the dependent_root changed. + let validators_to_update = { + // Avoid holding the read-lock for any longer than required. + let il_duties = duties_service.il_duties.read(); + let needs_update = il_duties.get(&epoch).is_none_or(|(prior_root, _duties)| { + // Update if dependent_root changed + *prior_root != dependent_root + }); + + if needs_update { + local_pubkeys.iter().collect::>() + } else { + Vec::new() + } + }; + + if validators_to_update.is_empty() { + // No validators have conflicting (epoch, dependent_root) values for this epoch. + return Ok(()); + } + + // Make a request for all indices that require updating which we have not already made a request for. + let indices_to_request = validators_to_update + .iter() + .filter_map(|pubkey| duties_service.validator_store.validator_index(pubkey)) + .filter(|validator_index| !initial_indices_to_request.contains(validator_index)) + .collect::>(); + + // Filter the initial duties by their relevance so that we don't hit warnings about + // overwriting duties. + let new_initial_duties = response + .data + .into_iter() + .filter(|duty| validators_to_update.contains(&&duty.pubkey)); + + let mut new_duties = if !indices_to_request.is_empty() { + post_validator_il_duties(duties_service, epoch, indices_to_request.as_slice()) + .await? + .data + } else { + vec![] + }; + new_duties.extend(new_initial_duties); + + drop(fetch_timer); + + let _store_timer = validator_metrics::start_timer_vec( + &validator_metrics::DUTIES_SERVICE_TIMES, + &[validator_metrics::UPDATE_IL_DUTIES_STORE], + ); + + debug!( + %dependent_root, + num_new_duties = new_duties.len(), + "Downloaded IL duties" + ); + + // Update duties - we only reach here if dependent_root changed or epoch is missing + let mut il_duties = duties_service.il_duties.write(); + + match il_duties.entry(epoch) { + hash_map::Entry::Occupied(mut entry) => { + // Dependent root must have changed, so we do complete replacement. + let (existing_root, _existing_duties) = entry.get(); + debug!( + old_root = %existing_root, + new_root = %dependent_root, + "IL dependent root changed, replacing all duties" + ); + + *entry.get_mut() = (dependent_root, new_duties); + } + hash_map::Entry::Vacant(entry) => { + // No existing duties for this epoch + entry.insert((dependent_root, new_duties)); + } + } + + Ok(()) +} + /// Notify the block service if it should produce a block. async fn notify_block_production_service( current_slot: Slot, diff --git a/validator_client/validator_services/src/notifier_service.rs b/validator_client/validator_services/src/notifier_service.rs index e6e7a678640..38ee32e7bdb 100644 --- a/validator_client/validator_services/src/notifier_service.rs +++ b/validator_client/validator_services/src/notifier_service.rs @@ -110,6 +110,7 @@ pub async fn notify( let proposing_validators = duties_service.proposer_count(epoch); let attesting_validators = duties_service.attester_count(epoch); let ptc_validators = duties_service.ptc_count(epoch); + let il_validators = duties_service.il_duties_count(epoch); let doppelganger_detecting_validators = duties_service.doppelganger_detecting_count(); if doppelganger_detecting_validators > 0 { @@ -128,6 +129,7 @@ pub async fn notify( info!( current_epoch_proposers = proposing_validators, current_epoch_ptc = ptc_validators, + current_epoch_il = il_validators, active_validators = attesting_validators, total_validators = total_validators, %epoch, @@ -138,6 +140,7 @@ pub async fn notify( info!( current_epoch_proposers = proposing_validators, current_epoch_ptc = ptc_validators, + current_epoch_il = il_validators, active_validators = attesting_validators, total_validators = total_validators, %epoch,