Add IL committee duties polling to the duties service

validator_client/http_metrics/src/lib.rs

@@ -205,6 +205,16 @@ pub fn gather_prometheus_metrics<E: EthSpec>(
                 &[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,
+            );
         }
     }
 

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<Result<IntGaugeVec>> = LazyLock::new(|| {
         &["task"],
     )
 });
+pub static IL_COUNT: LazyLock<Result<IntGaugeVec>> = 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<Result<IntCounter>> = LazyLock::new(|| {
     try_create_int_counter(
         "vc_beacon_block_proposal_changed",

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<T> {
     UnableToReadSlotClock,
     FailedToDownloadAttesters(#[allow(dead_code)] String),
     FailedToDownloadPtc(#[allow(dead_code)] String),
+    FailedToDownloadILs(#[allow(dead_code)] String),
     FailedToProduceSelectionProof(#[allow(dead_code)] ValidatorStoreError<T>),
     InvalidModulo(#[allow(dead_code)] ArithError),
     Arith(#[allow(dead_code)] ArithError),
@@ -286,6 +288,7 @@ type DependentRoot = Hash256;
 type AttesterMap = HashMap<PublicKeyBytes, HashMap<Epoch, (DependentRoot, DutyAndProof)>>;
 type ProposerMap = HashMap<Epoch, (DependentRoot, Vec<ProposerData>)>;
 type PtcMap = HashMap<Epoch, (DependentRoot, Vec<PtcDuty>)>;
+type IlMap = HashMap<Epoch, (DependentRoot, Vec<InclusionListDuty>)>;
 
 pub struct DutiesServiceBuilder<S, T> {
     /// Provides the canonical list of locally-managed validators.
@@ -395,6 +398,7 @@ impl<S, T> DutiesServiceBuilder<S, T> {
             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<S, T> {
     pub sync_duties: SyncDutiesMap,
     /// Maps an epoch to PTC duties for locally-managed validators.
     pub ptc_duties: RwLock<PtcMap>,
+    /// Maps an epoch to the IL committee duties for locally-managed validators.
+    pub il_duties: RwLock<IlMap>,
     /// Provides the canonical list of locally-managed validators.
     pub validator_store: Arc<S>,
     /// Maps unknown validator pubkeys to the next slot time when a poll should be conducted again.
@@ -500,6 +506,15 @@ impl<S: ValidatorStore, T: SlotClock + 'static> DutiesService<S, T> {
             .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
@@ -764,6 +779,54 @@ pub fn start_update_service<S: ValidatorStore + 'static, T: SlotClock + 'static>
             "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 +1467,26 @@ async fn post_validator_duties_ptc<S: ValidatorStore, T: SlotClock + 'static>(
         .map_err(|e| Error::FailedToDownloadPtc(e.to_string()))
 }
 
+async fn post_validator_il_duties<S: ValidatorStore, T: SlotClock + 'static>(
+    duties_service: &Arc<DutiesService<S, T>>,
+    epoch: Epoch,
+    validator_indices: &[u64],
+) -> Result<DutiesResponse<Vec<InclusionListDuty>>, Error<S::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 +2067,189 @@ async fn poll_beacon_ptc_attesters_for_epoch<
     Ok(())
 }
 
+async fn poll_beacon_il_committee_duties<S: ValidatorStore + 'static, T: SlotClock + 'static>(
+    duties_service: &Arc<DutiesService<S, T>>,
+) -> Result<(), Error<S::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<PublicKeyBytes> = 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::<Vec<_>>();
+
+    // 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<S: ValidatorStore + 'static, T: SlotClock + 'static>(
+    duties_service: &Arc<DutiesService<S, T>>,
+    epoch: Epoch,
+    local_indices: &[u64],
+    local_pubkeys: &HashSet<PublicKeyBytes>,
+) -> Result<(), Error<S::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::<Vec<_>>()
+        } 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::<Vec<_>>();
+
+    // 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<S: ValidatorStore>(
     current_slot: Slot,