async fn poll_beacon_proposers<T: SlotClock + 'static, E: EthSpec>(
    duties_service: &DutiesService<T, E>,
    block_service_tx: &mut Sender<BlockServiceNotification>,
) -> Result<(), Error> {
    let _timer =
        metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::UPDATE_PROPOSERS]);

    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());

    // Notify the block proposal service for any proposals that we have in our cache.
    //
    // See the function-level documentation for more information.
    let initial_block_proposers = duties_service.block_proposers(current_slot);
    notify_block_production_service(
        current_slot,
        &initial_block_proposers,
        block_service_tx,
        &log,
    )
    .await;

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

    // Only download duties and push out additional block production events if we have some
    // validators.
    if !local_pubkeys.is_empty() {
        let download_result = duties_service
            .beacon_nodes
            .first_success(duties_service.require_synced, |beacon_node| async move {
                beacon_node
                    .get_validator_duties_proposer(current_epoch)
                    .await
            })
            .await;

        match download_result {
            Ok(response) => {
                let dependent_root = response.dependent_root;

                let relevant_duties = response
                    .data
                    .into_iter()
                    .filter(|proposer_duty| local_pubkeys.contains(&proposer_duty.pubkey))
                    .collect::<Vec<_>>();

                debug!(
                    log,
                    "Downloaded proposer duties";
                    "dependent_root" => %dependent_root,
                    "num_relevant_duties" => relevant_duties.len(),
                );

                if let Some((prior_dependent_root, _)) = duties_service
                    .proposers
                    .write()
                    .insert(current_epoch, (dependent_root, relevant_duties))
                {
                    if dependent_root != prior_dependent_root {
                        warn!(
                            log,
                            "Proposer duties re-org";
                            "prior_dependent_root" => %prior_dependent_root,
                            "dependent_root" => %dependent_root,
                            "msg" => "this may happen from time to time"
                        )
                    }
                }
            }
            // Don't return early here, we still want to try and produce blocks using the cached values.
            Err(e) => error!(
                log,
                "Failed to download proposer duties";
                "err" => %e,
            ),
        }

        // Compute the block proposers for this slot again, now that we've received an update from
        // the BN.
        //
        // Then, compute the difference between these two sets to obtain a set of block proposers
        // which were not included in the initial notification to the `BlockService`.
        let additional_block_producers = duties_service
            .block_proposers(current_slot)
            .difference(&initial_block_proposers)
            .copied()
            .collect::<HashSet<PublicKeyBytes>>();

        // If there are any new proposers for this slot, send a notification so they produce a
        // block.
        //
        // See the function-level documentation for more reasoning about this behaviour.
        if !additional_block_producers.is_empty() {
            notify_block_production_service(
                current_slot,
                &additional_block_producers,
                block_service_tx,
                &log,
            )
            .await;
            debug!(
                log,
                "Detected new block proposer";
                "current_slot" => current_slot,
            );
            metrics::inc_counter(&metrics::PROPOSAL_CHANGED);
        }
    }

    // Prune old duties.
    duties_service
        .proposers
        .write()
        .retain(|&epoch, _| epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch);

    Ok(())
}
/// Download the proposer duties for the current epoch and store them in `duties_service.proposers`.
/// If there are any proposer for this slot, send out a notification to the block proposers.
///
/// ## Note
///
/// This function will potentially send *two* notifications to the `BlockService`; it will send a
/// notification initially, then it will download the latest duties and send a *second* notification
/// if those duties have changed. This behaviour simultaneously achieves the following:
///
/// 1. Block production can happen immediately and does not have to wait for the proposer duties to
///    download.
/// 2. We won't miss a block if the duties for the current slot happen to change with this poll.
///
/// This sounds great, but is it safe? Firstly, the additional notification will only contain block
/// producers that were not included in the first notification. This should be safety enough.
/// However, we also have the slashing protection as a second line of defence. These two factors
/// provide an acceptable level of safety.
///
/// It's important to note that since there is a 0-epoch look-ahead (i.e., no look-ahead) for block
/// proposers then it's very likely that a proposal for the first slot of the epoch will need go
/// through the slow path every time. I.e., the proposal will only happen after we've been able to
/// download and process the duties from the BN. This means it is very important to ensure this
/// function is as fast as possible.
Download the proposer duties for the current epoch and store them in `duties_service.proposers`.
If there are any proposer for this slot, send out a notification to the block proposers.
Download the proposer duties for the current epoch and store them in ` duties_service . proposers ` . If there are any proposer for this slot send out a notification to the block proposers .
async fn poll_beacon_proposers<T: SlotClock + 'static, E: EthSpec>(
    duties_service: &DutiesService<T, E>,
    block_service_tx: &mut Sender<BlockServiceNotification>,
) -> Result<(), Error> {
    let _timer =
        metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::UPDATE_PROPOSERS]);
    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());
    // Notify the block proposal service for any proposals that we have in our cache.
    //
    // See the function-level documentation for more information.
    let initial_block_proposers = duties_service.block_proposers(current_slot);
    notify_block_production_service(
        current_slot,
        &initial_block_proposers,
        block_service_tx,
        &log,
    )
    .await;
    let local_pubkeys: HashSet<PublicKeyBytes> = duties_service
        .validator_store
        .voting_pubkeys()
        .into_iter()
        .collect();
    // Only download duties and push out additional block production events if we have some
    // validators.
    if !local_pubkeys.is_empty() {
        let download_result = duties_service
            .beacon_nodes
            .first_success(duties_service.require_synced, |beacon_node| async move {
                beacon_node
                    .get_validator_duties_proposer(current_epoch)
                    .await
            })
            .await;
        match download_result {
            Ok(response) => {
                let dependent_root = response.dependent_root;
                let relevant_duties = response
                    .data
                    .into_iter()
                    .filter(|proposer_duty| local_pubkeys.contains(&proposer_duty.pubkey))
                    .collect::<Vec<_>>();
                debug!(
                    log,
                    "Downloaded proposer duties";
                    "dependent_root" => %dependent_root,
                    "num_relevant_duties" => relevant_duties.len(),
                );
                if let Some((prior_dependent_root, _)) = duties_service
                    .proposers
                    .write()
                    .insert(current_epoch, (dependent_root, relevant_duties))
                {
                    if dependent_root != prior_dependent_root {
                        warn!(
                            log,
                            "Proposer duties re-org";
                            "prior_dependent_root" => %prior_dependent_root,
                            "dependent_root" => %dependent_root,
                            "msg" => "this may happen from time to time"
                        )
                    }
                }
            }
            // Don't return early here, we still want to try and produce blocks using the cached values.
            Err(e) => error!(
                log,
                "Failed to download proposer duties";
                "err" => %e,
            ),
        }
        // Compute the block proposers for this slot again, now that we've received an update from
        // the BN.
        //
        // Then, compute the difference between these two sets to obtain a set of block proposers
        // which were not included in the initial notification to the `BlockService`.
        let additional_block_producers = duties_service
            .block_proposers(current_slot)
            .difference(&initial_block_proposers)
            .copied()
            .collect::<HashSet<PublicKeyBytes>>();
        // If there are any new proposers for this slot, send a notification so they produce a
        // block.
        //
        // See the function-level documentation for more reasoning about this behaviour.
        if !additional_block_producers.is_empty() {
            notify_block_production_service(
                current_slot,
                &additional_block_producers,
                block_service_tx,
                &log,
            )
            .await;
            debug!(
                log,
                "Detected new block proposer";
                "current_slot" => current_slot,
            );
            metrics::inc_counter(&metrics::PROPOSAL_CHANGED);
        }
    }
    // Prune old duties.
    duties_service
        .proposers
        .write()
        .retain(|&epoch, _| epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch);
    Ok(())
}
async fn poll_beacon_proposers < T : SlotClock + ' static , E : EthSpec > ( duties_service : & DutiesService < T , E > , block_service_tx : & mut Sender < BlockServiceNotification > , ) -> Result < ( ) , Error > { let _timer = metrics :: start_timer_vec ( & metrics :: DUTIES_SERVICE_TIMES , & [ metrics :: UPDATE_PROPOSERS ] ) ; 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 ( ) ) ; // Notify the block proposal service for any proposals that we have in our cache. // // See the function-level documentation for more information. let initial_block_proposers = duties_service . block_proposers ( current_slot ) ; notify_block_production_service ( current_slot , & initial_block_proposers , block_service_tx , & log , ) . await ; let local_pubkeys : HashSet < PublicKeyBytes > = duties_service . validator_store . voting_pubkeys ( ) . into_iter ( ) . collect ( ) ; // Only download duties and push out additional block production events if we have some // validators. if ! local_pubkeys . is_empty ( ) { let download_result = duties_service . beacon_nodes . first_success ( duties_service . require_synced , | beacon_node | async move { beacon_node . get_validator_duties_proposer ( current_epoch ) . await } ) . await ; match download_result { Ok ( response ) => { let dependent_root = response . dependent_root ; let relevant_duties = response . data . into_iter ( ) . filter ( | proposer_duty | local_pubkeys . contains ( & proposer_duty . pubkey ) ) . collect :: < Vec < _ > > ( ) ; debug ! ( log " " " " dependent_root " " relevant_duties len ( ) ) ; if let Some ( ( prior_dependent_root , _ ) ) = duties_service . proposers . write ( ) . insert ( current_epoch , ( dependent_root , relevant_duties ) ) { if dependent_root != prior_dependent_root { warn ! ( log " " " " prior_dependent_root " " dependent_root " " " " ) } } } // Don't return early here, we still want to try and produce blocks using the cached values. Err ( e ) => error ! ( log " " " " e ) , } // Compute the block proposers for this slot again, now that we've received an update from // the BN. // // Then, compute the difference between these two sets to obtain a set of block proposers // which were not included in the initial notification to the `BlockService`. let additional_block_producers = duties_service . block_proposers ( current_slot ) . difference ( & initial_block_proposers ) . copied ( ) . collect :: < HashSet < PublicKeyBytes > > ( ) ; // If there are any new proposers for this slot, send a notification so they produce a // block. // // See the function-level documentation for more reasoning about this behaviour. if ! additional_block_producers . is_empty ( ) { notify_block_production_service ( current_slot , & additional_block_producers , block_service_tx , & log , ) . await ; debug ! ( log " " " " current_slot ) ; metrics :: inc_counter ( & metrics :: PROPOSAL_CHANGED ) ; } } // Prune old duties. duties_service . proposers . write ( ) . retain ( | & epoch , _ | epoch + HISTORICAL_DUTIES_EPOCHS >= current_epoch ) ; Ok ( ( ) ) }
Download the proposer duties for the current epoch and store them in `duties_service.proposers`.
Download the proposer duties for the current epoch and store them in ` duties_service . proposers ` .
duties_service
&DutiesService<T, E>
block_service_tx
&mut Sender<BlockServiceNotification>
note
This function will potentially send *two* notifications to the `BlockService`; it will send a
notification initially, then it will download the latest duties and send a *second* notification
if those duties have changed. This behaviour simultaneously achieves the following.

1. Block production can happen immediately and does not have to wait for the proposer duties to
download.
2. We won't miss a block if the duties for the current slot happen to change with this poll.

This sounds great, but is it safe. Firstly, the additional notification will only contain block
producers that were not included in the first notification. This should be safety enough.
However, we also have the slashing protection as a second line of defence. These two factors
provide an acceptable level of safety.

It's important to note that since there is a 0-epoch look-ahead  for block
proposers then it's very likely that a proposal for the first slot of the epoch will need go
through the slow path every time. I.e., the proposal will only happen after we've been able to
download and process the duties from the BN. This means it is very important to ensure this
function is as fast as possible.
This function will potentially send * two * notifications to the ` BlockService ` ; it will send a notification initially then it will download the latest duties and send a * second * notification if those duties have changed . This behaviour simultaneously achieves the following . 1 . Block production can happen immediately and does not have to wait for the proposer duties to download . 2 . We won ' t miss a block if the duties for the current slot happen to change with this poll . This sounds great but is it safe . Firstly the additional notification will only contain block producers that were not included in the first notification . This should be safety enough . However we also have the slashing protection as a second line of defence . These two factors provide an acceptable level of safety . It ' s important to note that since there is a 0 - epoch look - ahead for block proposers then it ' s very likely that a proposal for the first slot of the epoch will need go through the slow path every time . I . e . the proposal will only happen after we ' ve been able to download and process the duties from the BN . This means it is very important to ensure this function is as fast as possible .