[Merged by Bors] - Detailed validator monitoring

beacon_node/beacon_chain/src/attestation_verification.rs

@@ -778,6 +778,11 @@ impl<T: BeaconChainTypes> VerifiedUnaggregatedAttestation<T> {
         &self.attestation
     }
 
+    /// Returns the wrapped `indexed_attestation`.
+    pub fn indexed_attestation(&self) -> &IndexedAttestation<T::EthSpec> {
+        &self.indexed_attestation
+    }
+
     /// Returns a mutable reference to the underlying attestation.
     ///
     /// Only use during testing since modifying the `IndexedAttestation` can cause the attestation

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -23,6 +23,9 @@ use crate::persisted_fork_choice::PersistedForkChoice;
 use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
 use crate::snapshot_cache::SnapshotCache;
 use crate::timeout_rw_lock::TimeoutRwLock;
+use crate::validator_monitor::{
+    ValidatorMonitor, HISTORIC_EPOCHS as VALIDATOR_MONITOR_HISTORIC_EPOCHS,
+};
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::BeaconForkChoiceStore;
 use crate::BeaconSnapshot;
@@ -242,6 +245,8 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub(crate) graffiti: Graffiti,
     /// Optional slasher.
     pub slasher: Option<Arc<Slasher<T::EthSpec>>>,
+    /// Provides monitoring of a set of explicitly defined validators.
+    pub validator_monitor: RwLock<ValidatorMonitor<T::EthSpec>>,
 }
 
 type BeaconBlockAndState<T> = (BeaconBlock<T>, BeaconState<T>);
@@ -1624,6 +1629,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 .map_err(|e| BlockError::BeaconChainError(e.into()))?;
         }
 
+        // Allow the validator monitor to learn about a new valid state.
+        self.validator_monitor
+            .write()
+            .process_valid_state(current_slot.epoch(T::EthSpec::slots_per_epoch()), &state);
+        let validator_monitor = self.validator_monitor.read();
+
         // Register each attestation in the block with the fork choice service.
         for attestation in &block.body.attestations[..] {
             let _fork_choice_attestation_timer =
@@ -1641,8 +1652,35 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 Err(ForkChoiceError::InvalidAttestation(_)) => Ok(()),
                 Err(e) => Err(BlockError::BeaconChainError(e.into())),
             }?;
+
+            // Only register this with the validator monitor when the block is sufficiently close to
+            // the current slot.
+            if VALIDATOR_MONITOR_HISTORIC_EPOCHS as u64 * T::EthSpec::slots_per_epoch()
+                + block.slot.as_u64()
+                >= current_slot.as_u64()
+            {
+                validator_monitor.register_attestation_in_block(
+                    &indexed_attestation,
+                    &block,
+                    &self.spec,
+                );
+            }
         }
 
+        for exit in &block.body.voluntary_exits {
+            validator_monitor.register_block_voluntary_exit(&exit.message)
+        }
+
+        for slashing in &block.body.attester_slashings {
+            validator_monitor.register_block_attester_slashing(slashing)
+        }
+
+        for slashing in &block.body.proposer_slashings {
+            validator_monitor.register_block_proposer_slashing(slashing)
+        }
+
+        drop(validator_monitor);
+
         metrics::observe(
             &metrics::OPERATIONS_PER_BLOCK_ATTESTATION,
             block.body.attestations.len() as f64,

beacon_node/beacon_chain/src/builder.rs

@@ -6,6 +6,7 @@ use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::shuffling_cache::ShufflingCache;
 use crate::snapshot_cache::{SnapshotCache, DEFAULT_SNAPSHOT_CACHE_SIZE};
 use crate::timeout_rw_lock::TimeoutRwLock;
+use crate::validator_monitor::ValidatorMonitor;
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::ChainConfig;
 use crate::{
@@ -26,8 +27,8 @@ use std::sync::Arc;
 use std::time::Duration;
 use store::{HotColdDB, ItemStore};
 use types::{
-    BeaconBlock, BeaconState, ChainSpec, EthSpec, Graffiti, Hash256, Signature, SignedBeaconBlock,
-    Slot,
+    BeaconBlock, BeaconState, ChainSpec, EthSpec, Graffiti, Hash256, PublicKeyBytes, Signature,
+    SignedBeaconBlock, Slot,
 };
 
 pub const PUBKEY_CACHE_FILENAME: &str = "pubkey_cache.ssz";
@@ -88,6 +89,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     log: Option<Logger>,
     graffiti: Graffiti,
     slasher: Option<Arc<Slasher<T::EthSpec>>>,
+    validator_monitor: Option<ValidatorMonitor<T::EthSpec>>,
 }
 
 impl<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>
@@ -126,6 +128,7 @@ where
             log: None,
             graffiti: Graffiti::default(),
             slasher: None,
+            validator_monitor: None,
         }
     }
 
@@ -170,8 +173,8 @@ where
     /// Sets the logger.
     ///
     /// Should generally be called early in the build chain.
-    pub fn logger(mut self, logger: Logger) -> Self {
-        self.log = Some(logger);
+    pub fn logger(mut self, log: Logger) -> Self {
+        self.log = Some(log);
         self
     }
 
@@ -391,6 +394,23 @@ where
         self
     }
 
+    /// Register some validators for additional monitoring.
+    ///
+    /// `validators` is a comma-separated string of 0x-formatted BLS pubkeys.
+    pub fn monitor_validators(
+        mut self,
+        auto_register: bool,
+        validators: Vec<PublicKeyBytes>,
+        log: Logger,
+    ) -> Self {
+        self.validator_monitor = Some(ValidatorMonitor::new(
+            validators,
+            auto_register,
+            log.clone(),
+        ));
+        self
+    }
+
     /// Consumes `self`, returning a `BeaconChain` if all required parameters have been supplied.
     ///
     /// An error will be returned at runtime if all required parameters have not been configured.
@@ -418,6 +438,9 @@ where
         let genesis_state_root = self
             .genesis_state_root
             .ok_or("Cannot build without a genesis state root")?;
+        let mut validator_monitor = self
+            .validator_monitor
+            .ok_or("Cannot build without a validator monitor")?;
 
         let current_slot = if slot_clock
             .is_prior_to_genesis()
@@ -496,6 +519,13 @@ where
             log.clone(),
         );
 
+        if let Some(slot) = slot_clock.now() {
+            validator_monitor.process_valid_state(
+                slot.epoch(TEthSpec::slots_per_epoch()),
+                &canonical_head.beacon_state,
+            );
+        }
+
         let beacon_chain = BeaconChain {
             spec: self.spec,
             config: self.chain_config,
@@ -538,6 +568,7 @@ where
             log: log.clone(),
             graffiti: self.graffiti,
             slasher: self.slasher.clone(),
+            validator_monitor: RwLock::new(validator_monitor),
         };
 
         let head = beacon_chain
@@ -706,6 +737,7 @@ mod test {
             .testing_slot_clock(Duration::from_secs(1))
             .expect("should configure testing slot clock")
             .shutdown_sender(shutdown_tx)
+            .monitor_validators(true, vec![], log.clone())
             .build()
             .expect("should build");
 

beacon_node/beacon_chain/src/lib.rs

@@ -23,6 +23,7 @@ mod shuffling_cache;
 mod snapshot_cache;
 pub mod test_utils;
 mod timeout_rw_lock;
+pub mod validator_monitor;
 mod validator_pubkey_cache;
 
 pub use self::beacon_chain::{