Optimize validator duties (#2243)

## Issue Addressed

Closes #2052

## Proposed Changes

- Refactor the attester/proposer duties endpoints in the BN
    - Performance improvements
    - Fixes some potential inconsistencies with the dependent root fields.
    - Removes `http_api::beacon_proposer_cache` and just uses the one on the `BeaconChain` instead.
    - Move the code for the proposer/attester duties endpoints into separate files, for readability.
- Refactor the `DutiesService` in the VC
    - Required to reduce the delay on broadcasting new blocks.
    - Gets rid of the `ValidatorDuty` shim struct that came about when we adopted the standard API.
    - Separate block/attestation duty tasks so that they don't block each other when one is slow.
- In the VC, use `PublicKeyBytes` to represent validators instead of `PublicKey`. `PublicKey` is a legit crypto object whilst `PublicKeyBytes` is just a byte-array, it's much faster to clone/hash `PublicKeyBytes` and this change has had a significant impact on runtimes.
    - Unfortunately this has created lots of dust changes.
 - In the BN, store `PublicKeyBytes` in the `beacon_proposer_cache` and allow access to them. The HTTP API always sends `PublicKeyBytes` over the wire and the conversion from `PublicKey` -> `PublickeyBytes` is non-trivial, especially when queries have 100s/1000s of validators (like Pyrmont).
 - Add the `state_processing::state_advance` mod which dedups a lot of the "apply `n` skip slots to the state" code.
    - This also fixes a bug with some functions which were failing to include a state root as per [this comment](072695284f/consensus/state_processing/src/state_advance.rs (L69-L74)). I couldn't find any instance of this bug that resulted in anything more severe than keying a shuffling cache by the wrong block root.
 - Swap the VC block service to use `mpsc` from `tokio` instead of `futures`. This is consistent with the rest of the code base.
    
~~This PR *reduces* the size of the codebase 🎉~~ It *used* to reduce the size of the code base before I added more comments. 

## Observations on Prymont

- Proposer duties times down from peaks of 450ms to consistent <1ms.
- Current epoch attester duties times down from >1s peaks to a consistent 20-30ms.
- Block production down from +600ms to 100-200ms.

## Additional Info

- ~~Blocked on #2241~~
- ~~Blocked on #2234~~

## TODO

- [x] ~~Refactor this into some smaller PRs?~~ Leaving this as-is for now.
- [x] Address `per_slot_processing` roots.
- [x] Investigate slow next epoch times. Not getting added to cache on block processing?
- [x] Consider [this](072695284f/beacon_node/store/src/hot_cold_store.rs (L811-L812)) in the scenario of replacing the state roots


Co-authored-by: pawan <pawandhananjay@gmail.com>
Co-authored-by: Michael Sproul <michael@sigmaprime.io>

beacon_node/beacon_chain/src/test_utils.rs

@@ -19,7 +19,7 @@ use rand::SeedableRng;
 use rayon::prelude::*;
 use slog::Logger;
 use slot_clock::TestingSlotClock;
-use state_processing::per_slot_processing;
+use state_processing::state_advance::complete_state_advance;
 use std::borrow::Cow;
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
@@ -331,6 +331,12 @@ where
         self.chain.head().unwrap().beacon_state
     }
 
+    pub fn get_current_state_and_root(&self) -> (BeaconState<E>, Hash256) {
+        let head = self.chain.head().unwrap();
+        let state_root = head.beacon_state_root();
+        (head.beacon_state, state_root)
+    }
+
     pub fn get_current_slot(&self) -> Slot {
         self.chain.slot().unwrap()
     }
@@ -377,10 +383,8 @@ where
         assert_ne!(slot, 0, "can't produce a block at slot 0");
         assert!(slot >= state.slot);
 
-        while state.slot < slot {
-            per_slot_processing(&mut state, None, &self.spec)
-                .expect("should be able to advance state to slot");
-        }
+        complete_state_advance(&mut state, None, slot, &self.spec)
+            .expect("should be able to advance state to slot");
 
         state
             .build_all_caches(&self.spec)
@@ -430,6 +434,7 @@ where
         &self,
         attesting_validators: &[usize],
         state: &BeaconState<E>,
+        state_root: Hash256,
         head_block_root: SignedBeaconBlockHash,
         attestation_slot: Slot,
     ) -> Vec<Vec<(Attestation<E>, SubnetId)>> {
@@ -454,6 +459,7 @@ where
                                 bc.index,
                                 head_block_root.into(),
                                 Cow::Borrowed(state),
+                                state_root,
                             )
                             .unwrap();
 
@@ -503,6 +509,7 @@ where
         &self,
         attestation_strategy: &AttestationStrategy,
         state: &BeaconState<E>,
+        state_root: Hash256,
         head_block_root: Hash256,
         attestation_slot: Slot,
     ) -> Vec<Vec<(Attestation<E>, SubnetId)>> {
@@ -513,6 +520,7 @@ where
         self.make_unaggregated_attestations(
             &validators,
             state,
+            state_root,
             head_block_root.into(),
             attestation_slot,
         )
@@ -522,11 +530,17 @@ where
         &self,
         attesting_validators: &[usize],
         state: &BeaconState<E>,
+        state_root: Hash256,
         block_hash: SignedBeaconBlockHash,
         slot: Slot,
     ) -> HarnessAttestations<E> {
-        let unaggregated_attestations =
-            self.make_unaggregated_attestations(&attesting_validators, &state, block_hash, slot);
+        let unaggregated_attestations = self.make_unaggregated_attestations(
+            &attesting_validators,
+            &state,
+            state_root,
+            block_hash,
+            slot,
+        );
 
         let aggregated_attestations: Vec<Option<SignedAggregateAndProof<E>>> = unaggregated_attestations
             .iter()
@@ -754,12 +768,18 @@ where
     pub fn attest_block(
         &self,
         state: &BeaconState<E>,
+        state_root: Hash256,
         block_hash: SignedBeaconBlockHash,
         block: &SignedBeaconBlock<E>,
         validators: &[usize],
     ) {
-        let attestations =
-            self.make_attestations(validators, &state, block_hash, block.message.slot);
+        let attestations = self.make_attestations(
+            validators,
+            &state,
+            state_root,
+            block_hash,
+            block.message.slot,
+        );
         self.process_attestations(attestations);
     }
 
@@ -767,26 +787,29 @@ where
         &self,
         slot: Slot,
         state: BeaconState<E>,
+        state_root: Hash256,
         validators: &[usize],
     ) -> Result<(SignedBeaconBlockHash, BeaconState<E>), BlockError<E>> {
         let (block_hash, block, state) = self.add_block_at_slot(slot, state)?;
-        self.attest_block(&state, block_hash, &block, validators);
+        self.attest_block(&state, state_root, block_hash, &block, validators);
         Ok((block_hash, state))
     }
 
     pub fn add_attested_blocks_at_slots(
         &self,
         state: BeaconState<E>,
+        state_root: Hash256,
         slots: &[Slot],
         validators: &[usize],
     ) -> AddBlocksResult<E> {
         assert!(!slots.is_empty());
-        self.add_attested_blocks_at_slots_given_lbh(state, slots, validators, None)
+        self.add_attested_blocks_at_slots_given_lbh(state, state_root, slots, validators, None)
     }
 
     fn add_attested_blocks_at_slots_given_lbh(
         &self,
         mut state: BeaconState<E>,
+        state_root: Hash256,
         slots: &[Slot],
         validators: &[usize],
         mut latest_block_hash: Option<SignedBeaconBlockHash>,
@@ -799,7 +822,7 @@ where
         let mut state_hash_from_slot: HashMap<Slot, BeaconStateHash> = HashMap::new();
         for slot in slots {
             let (block_hash, new_state) = self
-                .add_attested_block_at_slot(*slot, state, validators)
+                .add_attested_block_at_slot(*slot, state, state_root, validators)
                 .unwrap();
             state = new_state;
             block_hash_from_slot.insert(*slot, block_hash);
@@ -857,8 +880,14 @@ where
         for epoch in min_epoch.as_u64()..=max_epoch.as_u64() {
             let mut new_chains = vec![];
 
-            for (head_state, slots, validators, mut block_hashes, mut state_hashes, head_block) in
-                chains
+            for (
+                mut head_state,
+                slots,
+                validators,
+                mut block_hashes,
+                mut state_hashes,
+                head_block,
+            ) in chains
             {
                 let epoch_slots = slots
                     .iter()
@@ -866,9 +895,11 @@ where
                     .copied()
                     .collect::<Vec<_>>();
 
+                let head_state_root = head_state.update_tree_hash_cache().unwrap();
                 let (new_block_hashes, new_state_hashes, new_head_block, new_head_state) = self
                     .add_attested_blocks_at_slots_given_lbh(
                         head_state,
+                        head_state_root,
                         &epoch_slots,
                         &validators,
                         Some(head_block),
@@ -947,7 +978,7 @@ where
         block_strategy: BlockStrategy,
         attestation_strategy: AttestationStrategy,
     ) -> Hash256 {
-        let (state, slots) = match block_strategy {
+        let (mut state, slots) = match block_strategy {
             BlockStrategy::OnCanonicalHead => {
                 let current_slot: u64 = self.get_current_slot().into();
                 let slots: Vec<Slot> = (current_slot..(current_slot + (num_blocks as u64)))
@@ -975,8 +1006,9 @@ where
             AttestationStrategy::AllValidators => self.get_all_validators(),
             AttestationStrategy::SomeValidators(vals) => vals,
         };
+        let state_root = state.update_tree_hash_cache().unwrap();
         let (_, _, last_produced_block_hash, _) =
-            self.add_attested_blocks_at_slots(state, &slots, &validators);
+            self.add_attested_blocks_at_slots(state, state_root, &slots, &validators);
         last_produced_block_hash.into()
     }