async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
    duties_service: &DutiesService<T, E>,
) -> Result<(), Error> {
    let current_epoch_timer = metrics::start_timer_vec(
        &metrics::DUTIES_SERVICE_TIMES,
        &[metrics::UPDATE_ATTESTERS_CURRENT_EPOCH],
    );

    let log = duties_service.context.log();

    let current_slot = duties_service
        .slot_clock
        .now()
        .ok_or(Error::UnableToReadSlotClock)?;
    let current_epoch = current_slot.epoch(E::slots_per_epoch());
    let next_epoch = current_epoch + 1;

    let local_pubkeys: HashSet<PublicKeyBytes> = duties_service
        .validator_store
        .voting_pubkeys()
        .into_iter()
        .collect();

    let local_indices = {
        let mut local_indices = Vec::with_capacity(local_pubkeys.len());
        let indices_map = duties_service.indices.read();
        for &pubkey in &local_pubkeys {
            if let Some(validator_index) = indices_map.get(&pubkey) {
                local_indices.push(*validator_index)
            }
        }
        local_indices
    };

    // Download the duties and update the duties for the current epoch.
    if let Err(e) = poll_beacon_attesters_for_epoch(
        &duties_service,
        current_epoch,
        &local_indices,
        &local_pubkeys,
    )
    .await
    {
        error!(
            log,
            "Failed to download attester duties";
            "current_epoch" => current_epoch,
            "request_epoch" => current_epoch,
            "err" => ?e,
        )
    }

    drop(current_epoch_timer);
    let next_epoch_timer = metrics::start_timer_vec(
        &metrics::DUTIES_SERVICE_TIMES,
        &[metrics::UPDATE_ATTESTERS_NEXT_EPOCH],
    );

    // Download the duties and update the duties for the next epoch.
    if let Err(e) =
        poll_beacon_attesters_for_epoch(&duties_service, next_epoch, &local_indices, &local_pubkeys)
            .await
    {
        error!(
            log,
            "Failed to download attester duties";
            "current_epoch" => current_epoch,
            "request_epoch" => next_epoch,
            "err" => ?e,
        )
    }

    drop(next_epoch_timer);
    let subscriptions_timer =
        metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::SUBSCRIPTIONS]);

    // This vector is likely to be a little oversized, but it won't reallocate.
    let mut subscriptions = Vec::with_capacity(local_pubkeys.len() * 2);

    // For this epoch and the next epoch, produce any beacon committee subscriptions.
    //
    // We are *always* pushing out subscriptions, even if we've subscribed before. This is
    // potentially excessive on the BN in normal cases, but it will help with fast re-subscriptions
    // if the BN goes offline or we swap to a different one.
    for epoch in &[current_epoch, next_epoch] {
        duties_service
            .attesters
            .read()
            .iter()
            .filter_map(|(_, map)| map.get(&epoch))
            // The BN logs a warning if we try and subscribe to current or near-by slots. Give it a
            // buffer.
            .filter(|(_, duty_and_proof)| {
                current_slot + SUBSCRIPTION_BUFFER_SLOTS < duty_and_proof.duty.slot
            })
            .for_each(|(_, duty_and_proof)| {
                let duty = &duty_and_proof.duty;
                let is_aggregator = duty_and_proof.selection_proof.is_some();

                subscriptions.push(BeaconCommitteeSubscription {
                    validator_index: duty.validator_index,
                    committee_index: duty.committee_index,
                    committees_at_slot: duty.committees_at_slot,
                    slot: duty.slot,
                    is_aggregator,
                })
            });
    }

    // If there are any subscriptions, push them out to the beacon node.
    if !subscriptions.is_empty() {
        let subscriptions_ref = &subscriptions;
        if let Err(e) = duties_service
            .beacon_nodes
            .first_success(duties_service.require_synced, |beacon_node| async move {
                beacon_node
                    .post_validator_beacon_committee_subscriptions(subscriptions_ref)
                    .await
            })
            .await
        {
            error!(
                log,
                "Failed to subscribe validators";
                "error" => %e
            )
        }
    }

    drop(subscriptions_timer);

    // Prune old duties.
    duties_service
        .attesters
        .write()
        .iter_mut()
        .for_each(|(_, map)| {
            map.retain(|&epoch, _| epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch)
        });

    Ok(())
}
/// Query the beacon node for attestation duties for any known validators.
///
/// This function will perform (in the following order):
///
/// 1. Poll for current-epoch duties and update the local `duties_service.attesters` map.
/// 2. As above, but for the next-epoch.
/// 3. Push out any attestation subnet subscriptions to the BN.
/// 4. Prune old entries from `duties_service.attesters`.
Query the beacon node for attestation duties for any known validators.
This function will perform (in the following order).

1. Poll for current-epoch duties and update the local `duties_service.attesters` map.
2. As above, but for the next-epoch.
3. Push out any attestation subnet subscriptions to the BN.
4. Prune old entries from `duties_service.attesters`.
Query the beacon node for attestation duties for any known validators . This function will perform ( in the following order ) . 1 . Poll for current - epoch duties and update the local ` duties_service . attesters ` map . 2 . As above but for the next - epoch . 3 . Push out any attestation subnet subscriptions to the BN . 4 . Prune old entries from ` duties_service . attesters ` .
async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
    duties_service: &DutiesService<T, E>,
) -> Result<(), Error> {
    let current_epoch_timer = metrics::start_timer_vec(
        &metrics::DUTIES_SERVICE_TIMES,
        &[metrics::UPDATE_ATTESTERS_CURRENT_EPOCH],
    );
    let log = duties_service.context.log();
    let current_slot = duties_service
        .slot_clock
        .now()
        .ok_or(Error::UnableToReadSlotClock)?;
    let current_epoch = current_slot.epoch(E::slots_per_epoch());
    let next_epoch = current_epoch + 1;
    let local_pubkeys: HashSet<PublicKeyBytes> = duties_service
        .validator_store
        .voting_pubkeys()
        .into_iter()
        .collect();
    let local_indices = {
        let mut local_indices = Vec::with_capacity(local_pubkeys.len());
        let indices_map = duties_service.indices.read();
        for &pubkey in &local_pubkeys {
            if let Some(validator_index) = indices_map.get(&pubkey) {
                local_indices.push(*validator_index)
            }
        }
        local_indices
    };
    // Download the duties and update the duties for the current epoch.
    if let Err(e) = poll_beacon_attesters_for_epoch(
        &duties_service,
        current_epoch,
        &local_indices,
        &local_pubkeys,
    )
    .await
    {
        error!(
            log,
            "Failed to download attester duties";
            "current_epoch" => current_epoch,
            "request_epoch" => current_epoch,
            "err" => ?e,
        )
    }
    drop(current_epoch_timer);
    let next_epoch_timer = metrics::start_timer_vec(
        &metrics::DUTIES_SERVICE_TIMES,
        &[metrics::UPDATE_ATTESTERS_NEXT_EPOCH],
    );
    // Download the duties and update the duties for the next epoch.
    if let Err(e) =
        poll_beacon_attesters_for_epoch(&duties_service, next_epoch, &local_indices, &local_pubkeys)
            .await
    {
        error!(
            log,
            "Failed to download attester duties";
            "current_epoch" => current_epoch,
            "request_epoch" => next_epoch,
            "err" => ?e,
        )
    }
    drop(next_epoch_timer);
    let subscriptions_timer =
        metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::SUBSCRIPTIONS]);
    // This vector is likely to be a little oversized, but it won't reallocate.
    let mut subscriptions = Vec::with_capacity(local_pubkeys.len() * 2);
    // For this epoch and the next epoch, produce any beacon committee subscriptions.
    //
    // We are *always* pushing out subscriptions, even if we've subscribed before. This is
    // potentially excessive on the BN in normal cases, but it will help with fast re-subscriptions
    // if the BN goes offline or we swap to a different one.
    for epoch in &[current_epoch, next_epoch] {
        duties_service
            .attesters
            .read()
            .iter()
            .filter_map(|(_, map)| map.get(&epoch))
            // The BN logs a warning if we try and subscribe to current or near-by slots. Give it a
            // buffer.
            .filter(|(_, duty_and_proof)| {
                current_slot + SUBSCRIPTION_BUFFER_SLOTS < duty_and_proof.duty.slot
            })
            .for_each(|(_, duty_and_proof)| {
                let duty = &duty_and_proof.duty;
                let is_aggregator = duty_and_proof.selection_proof.is_some();
                subscriptions.push(BeaconCommitteeSubscription {
                    validator_index: duty.validator_index,
                    committee_index: duty.committee_index,
                    committees_at_slot: duty.committees_at_slot,
                    slot: duty.slot,
                    is_aggregator,
                })
            });
    }
    // If there are any subscriptions, push them out to the beacon node.
    if !subscriptions.is_empty() {
        let subscriptions_ref = &subscriptions;
        if let Err(e) = duties_service
            .beacon_nodes
            .first_success(duties_service.require_synced, |beacon_node| async move {
                beacon_node
                    .post_validator_beacon_committee_subscriptions(subscriptions_ref)
                    .await
            })
            .await
        {
            error!(
                log,
                "Failed to subscribe validators";
                "error" => %e
            )
        }
    }
    drop(subscriptions_timer);
    // Prune old duties.
    duties_service
        .attesters
        .write()
        .iter_mut()
        .for_each(|(_, map)| {
            map.retain(|&epoch, _| epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch)
        });
    Ok(())
}
async fn poll_beacon_attesters < T : SlotClock + ' static , E : EthSpec > ( duties_service : & DutiesService < T , E > , ) -> Result < ( ) , Error > { let current_epoch_timer = metrics :: start_timer_vec ( & metrics :: DUTIES_SERVICE_TIMES , & [ metrics :: UPDATE_ATTESTERS_CURRENT_EPOCH ] , ) ; let log = duties_service . context . log ( ) ; let current_slot = duties_service . slot_clock . now ( ) . ok_or ( Error :: UnableToReadSlotClock ) ? ; let current_epoch = current_slot . epoch ( E :: slots_per_epoch ( ) ) ; let next_epoch = current_epoch + 1 ; let local_pubkeys : HashSet < PublicKeyBytes > = duties_service . validator_store . voting_pubkeys ( ) . into_iter ( ) . collect ( ) ; let local_indices = { let mut local_indices = Vec :: with_capacity ( local_pubkeys . len ( ) ) ; let indices_map = duties_service . indices . read ( ) ; for & pubkey in & local_pubkeys { if let Some ( validator_index ) = indices_map . get ( & pubkey ) { local_indices . push ( * validator_index ) } } local_indices } ; // Download the duties and update the duties for the current epoch. if let Err ( e ) = poll_beacon_attesters_for_epoch ( & duties_service , current_epoch , & local_indices , & local_pubkeys , ) . await { error ! ( log " " " " current_epoch " " current_epoch " " e ) } drop ( current_epoch_timer ) ; let next_epoch_timer = metrics :: start_timer_vec ( & metrics :: DUTIES_SERVICE_TIMES , & [ metrics :: UPDATE_ATTESTERS_NEXT_EPOCH ] , ) ; // Download the duties and update the duties for the next epoch. if let Err ( e ) = poll_beacon_attesters_for_epoch ( & duties_service , next_epoch , & local_indices , & local_pubkeys ) . await { error ! ( log " " " " current_epoch " " next_epoch " " e ) } drop ( next_epoch_timer ) ; let subscriptions_timer = metrics :: start_timer_vec ( & metrics :: DUTIES_SERVICE_TIMES , & [ metrics :: SUBSCRIPTIONS ] ) ; // This vector is likely to be a little oversized, but it won't reallocate. let mut subscriptions = Vec :: with_capacity ( local_pubkeys . len ( ) * 2 ) ; // For this epoch and the next epoch, produce any beacon committee subscriptions. // // We are *always* pushing out subscriptions, even if we've subscribed before. This is // potentially excessive on the BN in normal cases, but it will help with fast re-subscriptions // if the BN goes offline or we swap to a different one. for epoch in & [ current_epoch , next_epoch ] { duties_service . attesters . read ( ) . iter ( ) . filter_map ( | ( _ , map ) | map . get ( & epoch ) ) // The BN logs a warning if we try and subscribe to current or near-by slots. Give it a // buffer. . filter ( | ( _ , duty_and_proof ) | { current_slot + SUBSCRIPTION_BUFFER_SLOTS < duty_and_proof . duty . slot } ) . for_each ( | ( _ , duty_and_proof ) | { let duty = & duty_and_proof . duty ; let is_aggregator = duty_and_proof . selection_proof . is_some ( ) ; subscriptions . push ( BeaconCommitteeSubscription { validator_index : duty . validator_index , committee_index : duty . committee_index , committees_at_slot : duty . committees_at_slot , slot : duty . slot , is_aggregator , } ) } ) ; } // If there are any subscriptions, push them out to the beacon node. if ! subscriptions . is_empty ( ) { let subscriptions_ref = & subscriptions ; if let Err ( e ) = duties_service . beacon_nodes . first_success ( duties_service . require_synced , | beacon_node | async move { beacon_node . post_validator_beacon_committee_subscriptions ( subscriptions_ref ) . await } ) . await { error ! ( log " " " " e ) } } drop ( subscriptions_timer ) ; // Prune old duties. duties_service . attesters . write ( ) . iter_mut ( ) . for_each ( | ( _ , map ) | { map . retain ( | & epoch , _ | epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch ) } ) ; Ok ( ( ) ) }
Query the beacon node for attestation duties for any known validators.
Query the beacon node for attestation duties for any known validators .
duties_service
&DutiesService<T, E>