From 119dc565a119f263bc5cf6fea72ac8d92b54c8d7 Mon Sep 17 00:00:00 2001 From: chonghe <44791194+chong-he@users.noreply.github.com> Date: Thu, 29 Jan 2026 14:44:17 +0800 Subject: [PATCH] Call beacon_committee_selections only once per epoch (#8699) * #8381 Add another if in the if/else branch for selections_endpoint case, and leaving the current code for non-selections_endpoint case mostly untouched. Now the if/else branch in `fill_in_selection_proofs` has 3 branches: - 1 for DVT with selections_endpoint (added in this PR) - 1 for DVT without selections_endpoint (Anchor) (untouched) - Non-DVT (untouched) Tested and it's working, also thanks very much to @KaloyanTanev for helping to test and confirmed that the calling has reduced and aggregated attestations are still working Co-Authored-By: Tan Chee Keong --- .../validator_services/src/duties_service.rs | 221 ++++++++++++------ 1 file changed, 149 insertions(+), 72 deletions(-) diff --git a/validator_client/validator_services/src/duties_service.rs b/validator_client/validator_services/src/duties_service.rs index c2378181ef..f467db92a1 100644 --- a/validator_client/validator_services/src/duties_service.rs +++ b/validator_client/validator_services/src/duties_service.rs @@ -17,7 +17,7 @@ use eth2::types::{ }; use futures::{ StreamExt, - stream::{self, FuturesUnordered}, + stream::{self, FuturesUnordered, TryStreamExt}, }; use parking_lot::{RwLock, RwLockWriteGuard}; use safe_arith::{ArithError, SafeArith}; @@ -141,71 +141,15 @@ impl Default for SelectionProofConfig { /// Create a selection proof for `duty`. /// /// Return `Ok(None)` if the attesting validator is not an aggregator. -async fn make_selection_proof( +async fn make_selection_proof( duty: &AttesterData, validator_store: &S, spec: &ChainSpec, - beacon_nodes: &Arc>, - config: &SelectionProofConfig, ) -> Result, Error> { - let selection_proof = if config.selections_endpoint { - let beacon_committee_selection = BeaconCommitteeSelection { - validator_index: duty.validator_index, - slot: duty.slot, - // This is partial selection proof - selection_proof: validator_store - .produce_selection_proof(duty.pubkey, duty.slot) - .await - .map_err(Error::FailedToProduceSelectionProof)? - .into(), - }; - // Call the endpoint /eth/v1/validator/beacon_committee_selections - // by sending the BeaconCommitteeSelection that contains partial selection proof - // The middleware should return BeaconCommitteeSelection that contains full selection proof - let middleware_response = beacon_nodes - .first_success(|beacon_node| { - let selection_data = beacon_committee_selection.clone(); - debug!( - "validator_index" = duty.validator_index, - "slot" = %duty.slot, - "partial selection proof" = ?beacon_committee_selection.selection_proof, - "Sending selection to middleware" - ); - async move { - beacon_node - .post_validator_beacon_committee_selections(&[selection_data]) - .await - } - }) - .await; - - let response_data = middleware_response - .map_err(|e| { - Error::FailedToProduceSelectionProof(ValidatorStoreError::Middleware(e.to_string())) - })? - .data - .pop() - .ok_or_else(|| { - Error::FailedToProduceSelectionProof(ValidatorStoreError::Middleware(format!( - "attestation selection proof - empty response for validator {}", - duty.validator_index - ))) - })?; - - debug!( - "validator_index" = response_data.validator_index, - "slot" = %response_data.slot, - // The selection proof from middleware response will be a full selection proof - "full selection proof" = ?response_data.selection_proof, - "Received selection from middleware" - ); - SelectionProof::from(response_data.selection_proof) - } else { - validator_store - .produce_selection_proof(duty.pubkey, duty.slot) - .await - .map_err(Error::FailedToProduceSelectionProof)? - }; + let selection_proof = validator_store + .produce_selection_proof(duty.pubkey, duty.slot) + .await + .map_err(Error::FailedToProduceSelectionProof)?; selection_proof .is_aggregator(duty.committee_length as usize, spec) @@ -221,6 +165,69 @@ async fn make_selection_proof( }) } +/// Create a Vec for every epoch +/// so that when calling the selections_endpoint later, it calls once per epoch with duties of all slots in that epoch +async fn make_beacon_committee_selection( + duties_service: &Arc>, + duties: &[AttesterData], +) -> Result, Error> { + // collect the BeaconCommitteeSelection in duties + let beacon_committee_selections = duties + .iter() + .map(|duty| { + let validator_store = &duties_service.validator_store; + async move { + let partial_selection_proof = validator_store + .produce_selection_proof(duty.pubkey, duty.slot) + .await + .map_err(Error::FailedToProduceSelectionProof)?; + Ok::>(BeaconCommitteeSelection { + validator_index: duty.validator_index, + slot: duty.slot, + selection_proof: partial_selection_proof.into(), + }) + } + }) + .collect::>() + .try_collect::>() + .await?; + + let epoch = duties + .first() + .map(|attester_data| attester_data.slot.epoch(S::E::slots_per_epoch())) + .unwrap_or_default(); + + debug!( + %epoch, + count = beacon_committee_selections.len(), + "Sending selections to middleware" + ); + + let selections = duties_service + .beacon_nodes + .first_success(|beacon_node| { + let selections = beacon_committee_selections.clone(); + async move { + beacon_node + .post_validator_beacon_committee_selections(&selections) + .await + } + }) + .await + .map_err(|e| { + Error::FailedToProduceSelectionProof(ValidatorStoreError::Middleware(e.to_string())) + })? + .data; + + debug!( + %epoch, + count = beacon_committee_selections.len(), + "Received selections from middleware" + ); + + Ok(selections) +} + impl DutyAndProof { /// Create a new `DutyAndProof` with the selection proof waiting to be filled in. pub fn new_without_selection_proof(duty: AttesterData, current_slot: Slot) -> Self { @@ -1287,14 +1294,21 @@ async fn fill_in_selection_proofs> = BTreeMap::new(); - for duty in duties { - duties_by_slot.entry(duty.slot).or_default().push(duty); + for duty in &duties { + duties_by_slot + .entry(duty.slot) + .or_default() + .push(duty.clone()); } // At halfway through each slot when nothing else is likely to be getting signed, sign a batch // of selection proofs and insert them into the duties service `attesters` map. let slot_clock = &duties_service.slot_clock; + // Create a HashMap for BeaconCommitteeSelection to match the duty later for distributed case involving middleware + let mut selection_hashmap = HashMap::new(); + let mut call_selection_endpoint = false; + while !duties_by_slot.is_empty() { if let Some(duration) = slot_clock.duration_to_next_slot() { sleep( @@ -1333,10 +1347,77 @@ async fn fill_in_selection_proofs selections, + Err(e) => { + error!( + error = ?e, + "Failed to fetch selection proofs" + ); + // If calling the endpoint fails, change to true so that it will retry the next slot + call_selection_endpoint = true; + continue; + } + }; + + for selection in &selections { + // This is a full_selection_proof returned by middleware + let selection_proof = + SelectionProof::from(selection.selection_proof.clone()); + selection_hashmap + .insert((selection.validator_index, selection.slot), selection_proof); + } + // Once we have the selection_proof, we don't call the selections_endpoint again + call_selection_endpoint = false; + } + + for duty in relevant_duties.into_values().flatten() { + let key = (duty.validator_index, duty.slot); + + let result = if let Some(selection_proof) = selection_hashmap.remove(&key) { + match selection_proof + .is_aggregator(duty.committee_length as usize, &duties_service.spec) + .map_err(Error::::InvalidModulo) + { + // Aggregator, return the result + Ok(true) => Ok((duty, Some(selection_proof))), + // Not an aggregator, do nothing and continue + Ok(false) => continue, + Err(_) => return, + } + } else { + Err(Error::FailedToProduceSelectionProof( + ValidatorStoreError::Middleware(format!( + "Missing selection proof for validator {} slot {}", + duty.validator_index, duty.slot + )), + )) + }; + + let mut attesters = duties_service.attesters.write(); + // if process_duty_and_proof returns false, exit the loop + if !process_duty_and_proof::( + &mut attesters, + result, + dependent_root, + current_slot, + ) { + return; + } + } + } + // For distributed case that uses parallel_sign + else if duties_service.selection_proof_config.parallel_sign { let mut duty_and_proof_results = relevant_duties .into_values() .flatten() @@ -1345,8 +1426,6 @@ async fn fill_in_selection_proofs