Add missed blocks to monitored validators

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3896,7 +3896,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             cached_head
                 .snapshot
                 .beacon_state
-                .proposer_shuffling_decision_root(proposer_head)?
+                .proposer_shuffling_decision_root(head_epoch, proposer_head)?
         } else {
             proposer_head
         };

beacon_node/beacon_chain/src/beacon_proposer_cache.rs

@@ -165,9 +165,10 @@ pub fn compute_proposer_duties_from_head<T: BeaconChainTypes>(
         .get_beacon_proposer_indices(&chain.spec)
         .map_err(BeaconChainError::from)?;
 
+
     let dependent_root = state
         // The only block which decides its own shuffling is the genesis block.
-        .proposer_shuffling_decision_root(chain.genesis_block_root)
+        .proposer_shuffling_decision_root(state.current_epoch(), chain.genesis_block_root)
         .map_err(BeaconChainError::from)?;
 
     Ok((indices, dependent_root, execution_status, state.fork()))

beacon_node/beacon_chain/src/canonical_head.rs

@@ -827,9 +827,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // These fields are used for server-sent events.
         let state_root = new_snapshot.beacon_state_root();
         let head_slot = new_snapshot.beacon_state.slot();
+        let head_epoch = head_slot.epoch(T::EthSpec::slots_per_epoch());
         let dependent_root = new_snapshot
             .beacon_state
-            .proposer_shuffling_decision_root(self.genesis_block_root);
+            .proposer_shuffling_decision_root(head_epoch, self.genesis_block_root);
         let prev_dependent_root = new_snapshot
             .beacon_state
             .attester_shuffling_decision_root(self.genesis_block_root, RelativeEpoch::Current);

beacon_node/beacon_chain/src/metrics.rs

@@ -813,6 +813,11 @@ lazy_static! {
         "Number of attester slashings seen",
         &["src", "validator"]
     );
+    pub static ref VALIDATOR_MONITOR_MISSED_NON_FINALIZED_BLOCKS_TOTAL: Result<IntCounterVec> = try_create_int_counter_vec(
+        "validator_monitor_missed_non_finalized_blocks_total",
+        "Number of non-finalized blocks missed",
+        &["epoch", "slot", "validator"]
+    );
 
     /*
      * Block Delay Metrics

beacon_node/beacon_chain/src/validator_monitor.rs

@@ -21,6 +21,7 @@ use types::{
     IndexedAttestation, ProposerSlashing, PublicKeyBytes, SignedAggregateAndProof,
     SignedContributionAndProof, Slot, SyncCommitteeMessage, VoluntaryExit,
 };
+use crate::beacon_proposer_cache::BeaconProposerCache;
 
 /// Used for Prometheus labels.
 ///
@@ -37,6 +38,9 @@ pub const HISTORIC_EPOCHS: usize = 10;
 /// Prometheus cardinality and log volume.
 pub const DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD: usize = 64;
 
+/// Lag slots used in detecting missed blocks for the monitored validators
+pub const MISSED_BLOCK_LAG_SLOTS: usize = 4;
+
 #[derive(Debug)]
 pub enum Error {
     InvalidPubkey(String),
@@ -343,7 +347,10 @@ pub struct ValidatorMonitor<T> {
     /// large validator counts causing infeasibly high cardinailty for
     /// Prometheus and high log volumes.
     individual_tracking_threshold: usize,
+    /// A Map representing the missed blocks by epoch, validator_index(state.validators) and slot
+    missed_blocks: HashSet<(Epoch, u64, Slot)>,
     log: Logger,
+    beacon_proposer_cache: BeaconProposerCache,
     _phantom: PhantomData<T>,
 }
 
@@ -352,14 +359,17 @@ impl<T: EthSpec> ValidatorMonitor<T> {
         pubkeys: Vec<PublicKeyBytes>,
         auto_register: bool,
         individual_tracking_threshold: usize,
+        beacon_proposer_cache: BeaconProposerCache,
         log: Logger,
     ) -> Self {
         let mut s = Self {
             validators: <_>::default(),
             indices: <_>::default(),
             auto_register,
             individual_tracking_threshold,
+            missed_blocks: <_>::default(),
             log,
+            beacon_proposer_cache,
             _phantom: PhantomData,
         };
         for pubkey in pubkeys {
@@ -411,6 +421,9 @@ impl<T: EthSpec> ValidatorMonitor<T> {
                 self.indices.insert(i, validator.pubkey);
             });
 
+        // Add missed blocks for the monitored validators
+        self.add_validators_missed_blocks(current_epoch, state);
+
         // Update metrics for individual validators.
         for monitored_validator in self.validators.values() {
             if let Some(i) = monitored_validator.index {
@@ -487,8 +500,66 @@ impl<T: EthSpec> ValidatorMonitor<T> {
                         u64_to_i64(validator.withdrawable_epoch),
                     );
                 }
+
+                // Add non finalized missed blocks for the monitored validators for the current slot of the current epoch
+                let current_slot = state.slot();
+                self.missed_blocks.contains(&(current_epoch, i as u64, current_slot))
+                    .then(|| {
+                        metrics::inc_counter_vec(
+                            &metrics::VALIDATOR_MONITOR_MISSED_NON_FINALIZED_BLOCKS_TOTAL,
+                            &[current_epoch.to_string().as_str(), current_slot.to_string().as_str(), id],
+                        );
+                    });
+            }
+        }
+    }
+
+    fn add_validators_missed_blocks(&mut self, current_epoch: Epoch, state: &BeaconState<T>) {
+        // Determine missed (non-finalized) blocks (can contain false positives)
+        let range_of_slot = (T::slots_per_epoch() as usize - MISSED_BLOCK_LAG_SLOTS) - 1;
+        for n in 0..range_of_slot {
+            let slot = state.slot() - Slot::new(n as u64);
+            let prev_slot = slot - 1;
+
+            // safeguard if we are on a new chain, the slot - 1 can be negative
+            if prev_slot < 0 {
+                break;
+            }
+
+            // condition for missed_block is defined such as block_root == block_root - n
+            // where the proposer who missed the block is the proposer of the block at block_root - n
+            if let (Ok(block_root), Ok(prev_block_root)) = (state.get_block_root(slot), state.get_block_root(prev_slot)) {
+                if block_root == prev_block_root {
+                    let epoch = slot.epoch(T::slots_per_epoch());
+                    if let Ok(shuffling_decision_block) = state.proposer_shuffling_decision_root(epoch, *block_root) {
+                        // TODO: AI(Joel) Ask the team if we can use the beacon proposer cache without the mutex
+                        // mutex is implemented in the caller (BeaconChain), not in the cache implementation
+                        if let Some(proposer) = self.beacon_proposer_cache.get_slot::<T>(shuffling_decision_block, slot) {
+                            // only add missed blocks for the proposer if it's in the list of monitored validators
+                            if self.validators
+                                .values()
+                                .into_iter()
+                                .find(|validator| validator.index == Some(proposer.index as u64))
+                                .is_some() {
+                                self.missed_blocks.insert((current_epoch, proposer.index as u64, slot));
+                            }
+                        } else {
+                            debug!(
+                                self.log,
+                                "Could not find proposer for a missed non-finalized block in beacon proposer cache";
+                                "slot" => slot,
+                                "block_root" => format!("{}", block_root),
+                                "shuffling_decision_block" => format!("{}", shuffling_decision_block),
+                            );
+                        }
+                    }
+                }
             }
         }
+
+        // Prune missed blocks that are prior to last finalized epoch
+        let finalized_epoch = state.finalized_checkpoint().epoch;
+        self.missed_blocks.retain(|(epoch, _, _)| *epoch >= finalized_epoch);
     }
 
     /// Run `func` with the `TOTAL_LABEL` and optionally the

beacon_node/http_api/src/proposer_duties.rs

@@ -97,12 +97,12 @@ fn try_proposer_duties_from_cache<T: BeaconChainTypes>(
     let head = chain.canonical_head.cached_head();
     let head_block = &head.snapshot.beacon_block;
     let head_block_root = head.head_block_root();
+    let head_epoch = head_block.slot().epoch(T::EthSpec::slots_per_epoch());
     let head_decision_root = head
         .snapshot
         .beacon_state
-        .proposer_shuffling_decision_root(head_block_root)
+        .proposer_shuffling_decision_root(head_epoch, head_block_root)
         .map_err(warp_utils::reject::beacon_state_error)?;
-    let head_epoch = head_block.slot().epoch(T::EthSpec::slots_per_epoch());
     let execution_optimistic = chain
         .is_optimistic_or_invalid_head_block(head_block)
         .map_err(warp_utils::reject::beacon_chain_error)?;
@@ -231,7 +231,7 @@ fn compute_historic_proposer_duties<T: BeaconChainTypes>(
     // We can supply the genesis block root as the block root since we know that the only block that
     // decides its own root is the genesis block.
     let dependent_root = state
-        .proposer_shuffling_decision_root(chain.genesis_block_root)
+        .proposer_shuffling_decision_root(state.current_epoch(), chain.genesis_block_root)
         .map_err(BeaconChainError::from)
         .map_err(warp_utils::reject::beacon_chain_error)?;
 

consensus/types/src/beacon_state.rs

@@ -613,15 +613,16 @@ impl<T: EthSpec> BeaconState<T> {
         cache.get_all_beacon_committees()
     }
 
-    /// Returns the block root which decided the proposer shuffling for the current epoch. This root
+    /// Returns the block root which decided the proposer shuffling for the epoch passed in parameter. This root
     /// can be used to key this proposer shuffling.
     ///
     /// ## Notes
     ///
     /// The `block_root` covers the one-off scenario where the genesis block decides its own
     /// shuffling. It should be set to the latest block applied to `self` or the genesis block root.
-    pub fn proposer_shuffling_decision_root(&self, block_root: Hash256) -> Result<Hash256, Error> {
-        let decision_slot = self.proposer_shuffling_decision_slot();
+    /// The `epoch` should be set to the epoch of the block root.
+    pub fn proposer_shuffling_decision_root(&self, epoch: Epoch, block_root: Hash256) -> Result<Hash256, Error> {
+        let decision_slot = self.proposer_shuffling_decision_slot(epoch);
         if self.slot() == decision_slot {
             Ok(block_root)
         } else {
@@ -631,8 +632,8 @@ impl<T: EthSpec> BeaconState<T> {
 
     /// Returns the slot at which the proposer shuffling was decided. The block root at this slot
     /// can be used to key the proposer shuffling for the current epoch.
-    fn proposer_shuffling_decision_slot(&self) -> Slot {
-        self.current_epoch()
+    fn proposer_shuffling_decision_slot(&self, epoch: Epoch) -> Slot {
+        epoch
             .start_slot(T::slots_per_epoch())
             .saturating_sub(1_u64)
     }