Add configurable block replayer (#2863)

## Issue Addressed

Successor to #2431

## Proposed Changes

* Add a `BlockReplayer` struct to abstract over the intricacies of calling `per_slot_processing` and `per_block_processing` while avoiding unnecessary tree hashing.
* Add a variant of the forwards state root iterator that does not require an `end_state`.
* Use the `BlockReplayer` when reconstructing states in the database. Use the efficient forwards iterator for frozen states.
* Refactor the iterators to remove `Arc<HotColdDB>` (this seems to be neater than making _everything_ an `Arc<HotColdDB>` as I did in #2431).

Supplying the state roots allow us to avoid building a tree hash cache at all when reconstructing historic states, which saves around 1 second flat (regardless of `slots-per-restore-point`). This is a small percentage of worst-case state load times with 200K validators and SPRP=2048 (~15s vs ~16s) but a significant speed-up for more frequent restore points: state loads with SPRP=32 should be now consistently <500ms instead of 1.5s (a ~3x speedup).

## Additional Info

Required by https://github.com/sigp/lighthouse/pull/2628

beacon_node/beacon_chain/tests/store_tests.rs

@@ -14,6 +14,7 @@ use lazy_static::lazy_static;
 use logging::test_logger;
 use maplit::hashset;
 use rand::Rng;
+use state_processing::BlockReplayer;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::convert::TryInto;
@@ -126,7 +127,7 @@ fn randomised_skips() {
         "head should be at the current slot"
     );
 
-    check_split_slot(&harness, store);
+    check_split_slot(&harness, store.clone());
     check_chain_dump(&harness, num_blocks_produced + 1);
     check_iterators(&harness);
 }
@@ -358,6 +359,191 @@ fn epoch_boundary_state_attestation_processing() {
     assert!(checked_pre_fin);
 }
 
+// Test that the `end_slot` for forwards block and state root iterators works correctly.
+#[test]
+fn forwards_iter_block_and_state_roots_until() {
+    let num_blocks_produced = E::slots_per_epoch() * 17;
+    let db_path = tempdir().unwrap();
+    let store = get_store(&db_path);
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+
+    let all_validators = &harness.get_all_validators();
+    let (mut head_state, mut head_state_root) = harness.get_current_state_and_root();
+    let head_block_root = harness.chain.head_info().unwrap().block_root;
+    let mut block_roots = vec![head_block_root];
+    let mut state_roots = vec![head_state_root];
+
+    for slot in (1..=num_blocks_produced).map(Slot::from) {
+        let (block_root, mut state) = harness
+            .add_attested_block_at_slot(slot, head_state, head_state_root, all_validators)
+            .unwrap();
+        head_state_root = state.update_tree_hash_cache().unwrap();
+        head_state = state;
+        block_roots.push(block_root.into());
+        state_roots.push(head_state_root);
+    }
+
+    check_finalization(&harness, num_blocks_produced);
+    check_split_slot(&harness, store.clone());
+
+    // The last restore point slot is the point at which the hybrid forwards iterator behaviour
+    // changes.
+    let last_restore_point_slot = store.get_latest_restore_point_slot();
+    assert!(last_restore_point_slot > 0);
+
+    let chain = &harness.chain;
+    let head_state = harness.get_current_state();
+    let head_slot = head_state.slot();
+    assert_eq!(head_slot, num_blocks_produced);
+
+    let test_range = |start_slot: Slot, end_slot: Slot| {
+        let mut block_root_iter = chain
+            .forwards_iter_block_roots_until(start_slot, end_slot)
+            .unwrap();
+        let mut state_root_iter = chain
+            .forwards_iter_state_roots_until(start_slot, end_slot)
+            .unwrap();
+
+        for slot in (start_slot.as_u64()..=end_slot.as_u64()).map(Slot::new) {
+            let block_root = block_roots[slot.as_usize()];
+            assert_eq!(block_root_iter.next().unwrap().unwrap(), (block_root, slot));
+
+            let state_root = state_roots[slot.as_usize()];
+            assert_eq!(state_root_iter.next().unwrap().unwrap(), (state_root, slot));
+        }
+    };
+
+    let split_slot = store.get_split_slot();
+    assert!(split_slot > last_restore_point_slot);
+
+    test_range(Slot::new(0), last_restore_point_slot);
+    test_range(last_restore_point_slot, last_restore_point_slot);
+    test_range(last_restore_point_slot - 1, last_restore_point_slot);
+    test_range(Slot::new(0), last_restore_point_slot - 1);
+    test_range(Slot::new(0), split_slot);
+    test_range(last_restore_point_slot - 1, split_slot);
+    test_range(Slot::new(0), head_state.slot());
+}
+
+#[test]
+fn block_replay_with_inaccurate_state_roots() {
+    let num_blocks_produced = E::slots_per_epoch() * 3 + 31;
+    let db_path = tempdir().unwrap();
+    let store = get_store(&db_path);
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+    let chain = &harness.chain;
+
+    harness.extend_chain(
+        num_blocks_produced as usize,
+        BlockStrategy::OnCanonicalHead,
+        AttestationStrategy::AllValidators,
+    );
+
+    // Slot must not be 0 mod 32 or else no blocks will be replayed.
+    let (mut head_state, head_root) = harness.get_current_state_and_root();
+    assert_ne!(head_state.slot() % 32, 0);
+
+    let mut fast_head_state = store
+        .get_inconsistent_state_for_attestation_verification_only(
+            &head_root,
+            Some(head_state.slot()),
+        )
+        .unwrap()
+        .unwrap();
+    assert_eq!(head_state.validators(), fast_head_state.validators());
+
+    head_state.build_all_committee_caches(&chain.spec).unwrap();
+    fast_head_state
+        .build_all_committee_caches(&chain.spec)
+        .unwrap();
+
+    assert_eq!(
+        head_state
+            .get_cached_active_validator_indices(RelativeEpoch::Current)
+            .unwrap(),
+        fast_head_state
+            .get_cached_active_validator_indices(RelativeEpoch::Current)
+            .unwrap()
+    );
+}
+
+#[test]
+fn block_replayer_hooks() {
+    let db_path = tempdir().unwrap();
+    let store = get_store(&db_path);
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+    let chain = &harness.chain;
+
+    let block_slots = vec![1, 3, 5, 10, 11, 12, 13, 14, 31, 32, 33]
+        .into_iter()
+        .map(Slot::new)
+        .collect::<Vec<_>>();
+    let max_slot = *block_slots.last().unwrap();
+    let all_slots = (0..=max_slot.as_u64()).map(Slot::new).collect::<Vec<_>>();
+
+    let (state, state_root) = harness.get_current_state_and_root();
+    let all_validators = harness.get_all_validators();
+    let (_, _, end_block_root, mut end_state) = harness.add_attested_blocks_at_slots(
+        state.clone(),
+        state_root,
+        &block_slots,
+        &all_validators,
+    );
+
+    let blocks = store
+        .load_blocks_to_replay(Slot::new(0), max_slot, end_block_root.into())
+        .unwrap();
+
+    let mut pre_slots = vec![];
+    let mut post_slots = vec![];
+    let mut pre_block_slots = vec![];
+    let mut post_block_slots = vec![];
+
+    let mut replay_state = BlockReplayer::<MinimalEthSpec>::new(state, &chain.spec)
+        .pre_slot_hook(Box::new(|state| {
+            pre_slots.push(state.slot());
+            Ok(())
+        }))
+        .post_slot_hook(Box::new(|state, epoch_summary, is_skip_slot| {
+            if is_skip_slot {
+                assert!(!block_slots.contains(&state.slot()));
+            } else {
+                assert!(block_slots.contains(&state.slot()));
+            }
+            if state.slot() % E::slots_per_epoch() == 0 {
+                assert!(epoch_summary.is_some());
+            }
+            post_slots.push(state.slot());
+            Ok(())
+        }))
+        .pre_block_hook(Box::new(|state, block| {
+            assert_eq!(state.slot(), block.slot());
+            pre_block_slots.push(block.slot());
+            Ok(())
+        }))
+        .post_block_hook(Box::new(|state, block| {
+            assert_eq!(state.slot(), block.slot());
+            post_block_slots.push(block.slot());
+            Ok(())
+        }))
+        .apply_blocks(blocks, None)
+        .unwrap()
+        .into_state();
+
+    // All but last slot seen by pre-slot hook
+    assert_eq!(&pre_slots, all_slots.split_last().unwrap().1);
+    // All but 0th slot seen by post-slot hook
+    assert_eq!(&post_slots, all_slots.split_first().unwrap().1);
+    // All blocks seen by both hooks
+    assert_eq!(pre_block_slots, block_slots);
+    assert_eq!(post_block_slots, block_slots);
+
+    // States match.
+    end_state.drop_all_caches().unwrap();
+    replay_state.drop_all_caches().unwrap();
+    assert_eq!(end_state, replay_state);
+}
+
 #[test]
 fn delete_blocks_and_states() {
     let db_path = tempdir().unwrap();
@@ -430,7 +616,7 @@ fn delete_blocks_and_states() {
     // Delete faulty fork
     // Attempting to load those states should find them unavailable
     for (state_root, slot) in
-        StateRootsIterator::new(store.clone(), &faulty_head_state).map(Result::unwrap)
+        StateRootsIterator::new(&store, &faulty_head_state).map(Result::unwrap)
     {
         if slot <= unforked_blocks {
             break;
@@ -441,7 +627,7 @@ fn delete_blocks_and_states() {
 
     // Double-deleting should also be OK (deleting non-existent things is fine)
     for (state_root, slot) in
-        StateRootsIterator::new(store.clone(), &faulty_head_state).map(Result::unwrap)
+        StateRootsIterator::new(&store, &faulty_head_state).map(Result::unwrap)
     {
         if slot <= unforked_blocks {
             break;
@@ -451,7 +637,7 @@ fn delete_blocks_and_states() {
 
     // Deleting the blocks from the fork should remove them completely
     for (block_root, slot) in
-        BlockRootsIterator::new(store.clone(), &faulty_head_state).map(Result::unwrap)
+        BlockRootsIterator::new(&store, &faulty_head_state).map(Result::unwrap)
     {
         if slot <= unforked_blocks + 1 {
             break;