Review feedback

beacon_node/beacon_chain/Cargo.toml

@@ -48,7 +48,7 @@ bls = { path = "../../crypto/bls" }
 safe_arith = { path = "../../consensus/safe_arith" }
 environment = { path = "../../lighthouse/environment" }
 bus = "2.2.3"
+itertools = "0.9.0"
 
 [dev-dependencies]
 lazy_static = "1.4.0"
-

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -23,6 +23,7 @@ use crate::snapshot_cache::SnapshotCache;
 use crate::timeout_rw_lock::TimeoutRwLock;
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::BeaconSnapshot;
+use itertools::process_results;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use slog::{crit, debug, error, info, trace, warn, Logger};
 use slot_clock::SlotClock;
@@ -322,7 +323,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn rev_iter_block_roots(
         &self,
     ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
-        let head = self.head().map_err(|e| Error::from(e))?;
+        let head = self.head()?;
         let iter = BlockRootsIterator::owned(self.store.clone(), head.beacon_state);
         Ok(
             std::iter::once(Ok((head.beacon_block_root, head.beacon_block.slot())))
@@ -398,12 +399,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ///     returned may be earlier than the wall-clock slot.
     pub fn rev_iter_state_roots(
         &self,
-    ) -> Result<Box<dyn Iterator<Item = Result<(Hash256, Slot), Error>>>, Error> {
+    ) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
         let head = self.head()?;
         let slot = head.beacon_state.slot;
         let iter = StateRootsIterator::owned(self.store.clone(), head.beacon_state);
         let iter = std::iter::once(Ok((head.beacon_state_root, slot))).chain(iter);
-        Ok(Box::new(iter.map(|result| result.map_err(|e| e.into()))))
+        Ok(Box::new(iter.map(|result| result.map_err(Into::into))))
     }
 
     /// Returns the block at the given slot, if any. Only returns blocks in the canonical chain.
@@ -568,19 +569,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 Ok(state)
             }
             Ordering::Less => {
-                let state_root = self
-                    .rev_iter_state_roots()?
-                    .take_while(|result| match result {
-                        Ok((_, current_slot)) => *current_slot >= slot,
-                        Err(_) => true,
-                    })
-                    .find(|result| match result {
-                        Ok((_, current_slot)) => *current_slot == slot,
-                        Err(_) => true,
-                    })
-                    .transpose()?
-                    .map(|(root, _slot)| root)
-                    .ok_or_else(|| Error::NoStateForSlot(slot))?;
+                let state_root = process_results(self.rev_iter_state_roots()?, |iter| {
+                    iter.take_while(|(_, current_slot)| *current_slot >= slot)
+                        .find(|(_, current_slot)| *current_slot == slot)
+                        .map(|(root, _slot)| root)
+                })?
+                .ok_or_else(|| Error::NoStateForSlot(slot))?;
 
                 Ok(self
                     .get_state(&state_root, Some(slot))?
@@ -655,14 +649,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ///
     /// Returns None if a block doesn't exist at the slot.
     pub fn root_at_slot(&self, target_slot: Slot) -> Result<Option<Hash256>, Error> {
-        Ok(self
-            .rev_iter_block_roots()?
-            .find(|result| match result {
-                Ok((_, slot)) => *slot == target_slot,
-                Err(_) => true,
-            })
-            .transpose()?
-            .map(|(root, _)| root))
+        process_results(self.rev_iter_state_roots()?, |mut iter| {
+            iter.find(|(_, slot)| *slot == target_slot)
+                .map(|(root, _)| root)
+        })
     }
 
     /// Returns the block proposer for a given slot.

beacon_node/store/src/errors.rs

@@ -15,6 +15,7 @@ pub enum Error {
     DBError { message: String },
     RlpError(String),
     BlockNotFound(Hash256),
+    NoContinuationData,
 }
 
 impl From<DecodeError> for Error {

beacon_node/store/src/forwards_iter.rs

@@ -1,6 +1,6 @@
 use crate::chunked_iter::ChunkedVectorIter;
 use crate::chunked_vector::BlockRoots;
-use crate::errors::Result;
+use crate::errors::{Error, Result};
 use crate::iter::BlockRootsIterator;
 use crate::{HotColdDB, Store};
 use std::sync::Arc;
@@ -72,8 +72,7 @@ impl SimpleForwardsBlockRootsIterator {
                 Err(_) => true,
             })
             .collect::<Result<Vec<_>>>()?;
-        let result = Self { values: values };
-        Ok(result)
+        Ok(Self { values: values })
     }
 }
 
@@ -136,9 +135,8 @@ impl<E: EthSpec> HybridForwardsBlockRootsIterator<E> {
                     // to a post-finalization iterator beginning from the last slot
                     // of the pre iterator.
                     None => {
-                        let (end_state, end_block_root) = continuation_data
-                            .take()
-                            .expect("HybridForwardsBlockRootsIterator: logic error");
+                        let (end_state, end_block_root) =
+                            continuation_data.take().ok_or(Error::NoContinuationData)?;
 
                         *self = PostFinalization {
                             iter: SimpleForwardsBlockRootsIterator::new(

beacon_node/store/src/iter.rs

@@ -182,19 +182,15 @@ impl<'a, T: EthSpec, U: Store<T>> RootsIterator<'a, T, U> {
             (Ok(block_root), Ok(state_root)) => Ok(Some((*block_root, *state_root, self.slot))),
             (Err(BeaconStateError::SlotOutOfBounds), Err(BeaconStateError::SlotOutOfBounds)) => {
                 // Read a `BeaconState` from the store that has access to prior historical roots.
-                let maybe_beacon_state =
+                let beacon_state =
                     next_historical_root_backtrack_state(&*self.store, &self.beacon_state)?;
 
-                if let Some(beacon_state) = maybe_beacon_state {
-                    self.beacon_state = Cow::Owned(beacon_state);
+                self.beacon_state = Cow::Owned(beacon_state);
 
-                    let block_root = *self.beacon_state.get_block_root(self.slot)?;
-                    let state_root = *self.beacon_state.get_state_root(self.slot)?;
+                let block_root = *self.beacon_state.get_block_root(self.slot)?;
+                let state_root = *self.beacon_state.get_state_root(self.slot)?;
 
-                    Ok(Some((block_root, state_root, self.slot)))
-                } else {
-                    Ok(None)
-                }
+                Ok(Some((block_root, state_root, self.slot)))
             }
             (Err(e), _) => Err(e.into()),
             (Ok(_), Err(e)) => Err(e.into()),
@@ -295,14 +291,16 @@ impl<'a, T: EthSpec, U: Store<T>> Iterator for BlockIterator<'a, T, U> {
 fn next_historical_root_backtrack_state<E: EthSpec, S: Store<E>>(
     store: &S,
     current_state: &BeaconState<E>,
-) -> Result<Option<BeaconState<E>>, Error> {
+) -> Result<BeaconState<E>, Error> {
     // For compatibility with the freezer database's restore points, we load a state at
     // a restore point slot (thus avoiding replaying blocks). In the case where we're
     // not frozen, this just means we might not jump back by the maximum amount on
     // our first jump (i.e. at most 1 extra state load).
     let new_state_slot = slot_of_prev_restore_point::<E>(current_state.slot);
     let new_state_root = current_state.get_state_root(new_state_slot)?;
-    store.get_state(new_state_root, Some(new_state_slot))
+    Ok(store
+        .get_state(new_state_root, Some(new_state_slot))?
+        .ok_or_else(|| BeaconStateError::MissingBeaconState((*new_state_root).into()))?)
 }
 
 /// Compute the slot of the last guaranteed restore point in the freezer database.